1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2
 
2
 
 
 
 
2
 
2
 
2
2
2
 
 
 
 
1
 
 
 
 
 
 
2
 
 
 
2
 
 
 
2
 
2
 
 
 
 
 
 
 
1
2
 
 
 
 
 
2
 
 
2
 
 
2
2
 
1
 
 
 
2
 
 
1
1
1
 
1
1
 
 
1
1
 
1
 
 
1
 
2
 
 
 
 
 
1
 
1
 
1
 
 
 
 
1
2
2
3
3
3
3
1
 
 
3
3
 
 
 
 
3
 
 
 
1
3
1
 
 
 
1
1
 
 
 
 
 
 
 
1
 
1
1
1
 
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
1
 
 
 
 
 
 
 
 
 
 
 
1
 
 
6
6
 
1
 
2
 
1
 
2
 
 
 
 
1
1
 
1
 
 
 
 
1
 
 
 
 
 
 
1
 
 
1
1
 
 
1
1
 
1
1
1
1
1
1
1
 
1
1
 
1
 
 
 
 
/*
example.js
 
quickstart example
 
instruction
    1. save this script as example.js
    2. run the shell command:
        $ npm install npmtest-pomelo && PORT=8081 node example.js
    3. play with the browser-demo on http://127.0.0.1:8081
*/
 
 
 
/* istanbul instrument in package npmtest_pomelo */
/*jslint
    bitwise: true,
    browser: true,
    maxerr: 8,
    maxlen: 96,
    node: true,
    nomen: true,
    regexp: true,
    stupid: true
*/
(function () {
    'use strict';
    var local;
 
 
 
    // run shared js-env code - pre-init
    (function () {
        // init local
        local = {};
        // init modeJs
        local.modeJs = (function () {
            try {
                return typeof navigator.userAgent === 'string' &&
                    typeof document.querySelector('body') === 'object' &&
                    typeof XMLHttpRequest.prototype.open === 'function' &&
                    'browser';
            } catch (errorCaughtBrowser) {
                return module.exports &&
                    typeof process.versions.node === 'string' &&
                    typeof require('http').createServer === 'function' &&
                    'node';
            }
        }());
        // init global
        local.global = local.modeJs === 'browser'
            ? window
            : global;
        // init utility2_rollup
        local = local.global.utility2_rollup || (local.modeJs === 'browser'
            ? local.global.utility2_npmtest_pomelo
            : global.utility2_moduleExports);
        // export local
        local.global.local = local;
    }());
    switch (local.modeJs) {
 
 
 
    // post-init
    // run browser js-env code - post-init
    /* istanbul ignore next */
    case 'browser':
        local.testRunBrowser = function (event) {
            Eif (!event || (event &&
                    event.currentTarget &&
                    event.currentTarget.className &&
                    event.currentTarget.className.includes &&
                    event.currentTarget.className.includes('onreset'))) {
                // reset output
                Array.from(
                    document.querySelectorAll('body > .resettable')
                ).forEach(function (element) {
                    switch (element.tagName) {
                    case 'INPUT':
                    case 'TEXTAREA':
                        element.value = '';
                        break;
                    default:
                        element.textContent = '';
                    }
                });
            }
            switch (event && event.currentTarget && event.currentTarget.id) {
            case 'testRunButton1':
                // show tests
                Eif (document.querySelector('#testReportDiv1').style.display === 'none') {
                    document.querySelector('#testReportDiv1').style.display = 'block';
                    document.querySelector('#testRunButton1').textContent =
                        'hide internal test';
                    local.modeTest = true;
                    local.testRunDefault(local);
                // hide tests
                } else {
                    document.querySelector('#testReportDiv1').style.display = 'none';
                    document.querySelector('#testRunButton1').textContent = 'run internal test';
                }
                break;
            // custom-case
            default:
                break;
            }
            Iif (document.querySelector('#inputTextareaEval1') && (!event || (event &&
                    event.currentTarget &&
                    event.currentTarget.className &&
                    event.currentTarget.className.includes &&
                    event.currentTarget.className.includes('oneval')))) {
                // try to eval input-code
                try {
                    /*jslint evil: true*/
                    eval(document.querySelector('#inputTextareaEval1').value);
                } catch (errorCaught) {
                    console.error(errorCaught);
                }
            }
        };
        // log stderr and stdout to #outputTextareaStdout1
        ['error', 'log'].forEach(function (key) {
            console[key + '_original'] = console[key];
            console[key] = function () {
                var element;
                console[key + '_original'].apply(console, arguments);
                element = document.querySelector('#outputTextareaStdout1');
                Iif (!element) {
                    return;
                }
                // append text to #outputTextareaStdout1
                element.value += Array.from(arguments).map(function (arg) {
                    return typeof arg === 'string'
                        ? arg
                        : JSON.stringify(arg, null, 4);
                }).join(' ') + '\n';
                // scroll textarea to bottom
                element.scrollTop = element.scrollHeight;
            };
        });
        // init event-handling
        ['change', 'click', 'keyup'].forEach(function (event) {
            Array.from(document.querySelectorAll('.on' + event)).forEach(function (element) {
                element.addEventListener(event, local.testRunBrowser);
            });
        });
        // run tests
        local.testRunBrowser();
        break;
 
 
 
    // run node js-env code - post-init
    /* istanbul ignore next */
    case 'node':
        // export local
        module.exports = local;
        // require modules
        local.fs = require('fs');
        local.http = require('http');
        local.url = require('url');
        // init assets
        local.assetsDict = local.assetsDict || {};
        /* jslint-ignore-begin */
        local.assetsDict['/assets.index.template.html'] = '\
<!doctype html>\n\
<html lang="en">\n\
<head>\n\
<meta charset="UTF-8">\n\
<meta name="viewport" content="width=device-width, initial-scale=1">\n\
<title>{{env.npm_package_name}} (v{{env.npm_package_version}})</title>\n\
<style>\n\
/*csslint\n\
    box-sizing: false,\n\
    universal-selector: false\n\
*/\n\
* {\n\
    box-sizing: border-box;\n\
}\n\
body {\n\
    background: #dde;\n\
    font-family: Arial, Helvetica, sans-serif;\n\
    margin: 2rem;\n\
}\n\
body > * {\n\
    margin-bottom: 1rem;\n\
}\n\
.utility2FooterDiv {\n\
    margin-top: 20px;\n\
    text-align: center;\n\
}\n\
</style>\n\
<style>\n\
/*csslint\n\
*/\n\
textarea {\n\
    font-family: monospace;\n\
    height: 10rem;\n\
    width: 100%;\n\
}\n\
textarea[readonly] {\n\
    background: #ddd;\n\
}\n\
</style>\n\
</head>\n\
<body>\n\
<!-- utility2-comment\n\
<div id="ajaxProgressDiv1" style="background: #d00; height: 2px; left: 0; margin: 0; padding: 0; position: fixed; top: 0; transition: background 0.5s, width 1.5s; width: 25%;"></div>\n\
utility2-comment -->\n\
<h1>\n\
<!-- utility2-comment\n\
    <a\n\
        {{#if env.npm_package_homepage}}\n\
        href="{{env.npm_package_homepage}}"\n\
        {{/if env.npm_package_homepage}}\n\
        target="_blank"\n\
    >\n\
utility2-comment -->\n\
        {{env.npm_package_name}} (v{{env.npm_package_version}})\n\
<!-- utility2-comment\n\
    </a>\n\
utility2-comment -->\n\
</h1>\n\
<h3>{{env.npm_package_description}}</h3>\n\
<!-- utility2-comment\n\
<h4><a download href="assets.app.js">download standalone app</a></h4>\n\
<button class="onclick onreset" id="testRunButton1">run internal test</button><br>\n\
<div id="testReportDiv1" style="display: none;"></div>\n\
utility2-comment -->\n\
\n\
\n\
\n\
<label>stderr and stdout</label>\n\
<textarea class="resettable" id="outputTextareaStdout1" readonly></textarea>\n\
<!-- utility2-comment\n\
{{#if isRollup}}\n\
<script src="assets.app.js"></script>\n\
{{#unless isRollup}}\n\
utility2-comment -->\n\
<script src="assets.utility2.rollup.js"></script>\n\
<script src="jsonp.utility2._stateInit?callback=window.utility2._stateInit"></script>\n\
<script src="assets.npmtest_pomelo.rollup.js"></script>\n\
<script src="assets.example.js"></script>\n\
<script src="assets.test.js"></script>\n\
<!-- utility2-comment\n\
{{/if isRollup}}\n\
utility2-comment -->\n\
<div class="utility2FooterDiv">\n\
    [ this app was created with\n\
    <a href="https://github.com/kaizhu256/node-utility2" target="_blank">utility2</a>\n\
    ]\n\
</div>\n\
</body>\n\
</html>\n\
';
        /* jslint-ignore-end */
        Iif (local.templateRender) {
            local.assetsDict['/'] = local.templateRender(
                local.assetsDict['/assets.index.template.html'],
                {
                    env: local.objectSetDefault(local.env, {
                        npm_package_description: 'the greatest app in the world!',
                        npm_package_name: 'my-app',
                        npm_package_nameAlias: 'my_app',
                        npm_package_version: '0.0.1'
                    })
                }
            );
        } else {
            local.assetsDict['/'] = local.assetsDict['/assets.index.template.html']
                .replace((/\{\{env\.(\w+?)\}\}/g), function (match0, match1) {
                    // jslint-hack
                    String(match0);
                    switch (match1) {
                    case 'npm_package_description':
                        return 'the greatest app in the world!';
                    case 'npm_package_name':
                        return 'my-app';
                    case 'npm_package_nameAlias':
                        return 'my_app';
                    case 'npm_package_version':
                        return '0.0.1';
                    }
                });
        }
        // run the cli
        Eif (local.global.utility2_rollup || module !== require.main) {
            break;
        }
        local.assetsDict['/assets.example.js'] =
            local.assetsDict['/assets.example.js'] ||
            local.fs.readFileSync(__filename, 'utf8');
        // bug-workaround - long $npm_package_buildCustomOrg
        /* jslint-ignore-begin */
        local.assetsDict['/assets.npmtest_pomelo.rollup.js'] =
            local.assetsDict['/assets.npmtest_pomelo.rollup.js'] ||
            local.fs.readFileSync(
                local.npmtest_pomelo.__dirname + '/lib.npmtest_pomelo.js',
                'utf8'
            ).replace((/^#!/), '//');
        /* jslint-ignore-end */
        local.assetsDict['/favicon.ico'] = local.assetsDict['/favicon.ico'] || '';
        // if $npm_config_timeout_exit exists,
        // then exit this process after $npm_config_timeout_exit ms
        if (Number(process.env.npm_config_timeout_exit)) {
            setTimeout(process.exit, Number(process.env.npm_config_timeout_exit));
        }
        // start server
        if (local.global.utility2_serverHttp1) {
            break;
        }
        process.env.PORT = process.env.PORT || '8081';
        console.error('server starting on port ' + process.env.PORT);
        local.http.createServer(function (request, response) {
            request.urlParsed = local.url.parse(request.url);
            if (local.assetsDict[request.urlParsed.pathname] !== undefined) {
                response.end(local.assetsDict[request.urlParsed.pathname]);
                return;
            }
            response.statusCode = 404;
            response.end();
        }).listen(process.env.PORT);
        break;
    }
}());
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
 
 
 
 
 
 
 
 
 
 
2
 
2
 
 
 
 
2
 
2
 
2
2
2
 
 
 
 
1
 
 
 
 
 
 
2
 
 
 
2
 
2
 
2
1
 
1
1
1
 
 
 
 
 
/* istanbul instrument in package npmtest_pomelo */
/*jslint
    bitwise: true,
    browser: true,
    maxerr: 8,
    maxlen: 96,
    node: true,
    nomen: true,
    regexp: true,
    stupid: true
*/
(function () {
    'use strict';
    var local;
 
 
 
    // run shared js-env code - pre-init
    (function () {
        // init local
        local = {};
        // init modeJs
        local.modeJs = (function () {
            try {
                return typeof navigator.userAgent === 'string' &&
                    typeof document.querySelector('body') === 'object' &&
                    typeof XMLHttpRequest.prototype.open === 'function' &&
                    'browser';
            } catch (errorCaughtBrowser) {
                return module.exports &&
                    typeof process.versions.node === 'string' &&
                    typeof require('http').createServer === 'function' &&
                    'node';
            }
        }());
        // init global
        local.global = local.modeJs === 'browser'
            ? window
            : global;
        // init utility2_rollup
        local = local.global.utility2_rollup || local;
        // init lib
        local.local = local.npmtest_pomelo = local;
        // init exports
        if (local.modeJs === 'browser') {
            local.global.utility2_npmtest_pomelo = local;
        } else {
            module.exports = local;
            module.exports.__dirname = __dirname;
            module.exports.module = module;
        }
    }());
}());
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
 
 
 
 
 
 
 
 
 
 
2
 
2
 
 
 
 
2
 
2
 
2
2
2
 
 
 
 
1
 
 
 
 
 
 
2
 
 
2
 
 
1
 
 
 
1
 
 
1
 
1
 
 
2
 
 
 
 
 
2
2
 
2
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
2
2
 
2
 
 
 
 
 
1
 
 
 
 
 
 
1
 
 
 
1
 
 
1
 
 
 
 
 
 
1
 
 
 
 
 
 
1
1
 
 
1
 
 
 
 
 
 
1
1
1
1
1
1
 
 
1
 
 
 
 
2
2
 
 
1
 
 
 
 
 
 
2
2
 
 
1
 
 
 
 
 
 
2
2
 
 
1
 
 
 
 
 
 
2
2
 
 
1
 
 
 
 
 
 
1
1
 
 
 
1
1
 
 
 
 
/* istanbul instrument in package npmtest_pomelo */
/*jslint
    bitwise: true,
    browser: true,
    maxerr: 8,
    maxlen: 96,
    node: true,
    nomen: true,
    regexp: true,
    stupid: true
*/
(function () {
    'use strict';
    var local;
 
 
 
    // run shared js-env code - pre-init
    (function () {
        // init local
        local = {};
        // init modeJs
        local.modeJs = (function () {
            try {
                return typeof navigator.userAgent === 'string' &&
                    typeof document.querySelector('body') === 'object' &&
                    typeof XMLHttpRequest.prototype.open === 'function' &&
                    'browser';
            } catch (errorCaughtBrowser) {
                return module.exports &&
                    typeof process.versions.node === 'string' &&
                    typeof require('http').createServer === 'function' &&
                    'node';
            }
        }());
        // init global
        local.global = local.modeJs === 'browser'
            ? window
            : global;
        switch (local.modeJs) {
        // re-init local from window.local
        case 'browser':
            local = local.global.utility2.objectSetDefault(
                local.global.utility2_rollup || local.global.local,
                local.global.utility2
            );
            break;
        // re-init local from example.js
        case 'node':
            local = (local.global.utility2_rollup || require('utility2'))
                .requireExampleJsFromReadme();
            break;
        }
        // export local
        local.global.local = local;
    }());
 
 
 
    // run shared js-env code - function
    (function () {
        return;
    }());
    switch (local.modeJs) {
 
 
 
    // run browser js-env code - function
    case 'browser':
        break;
 
 
 
    // run node js-env code - function
    case 'node':
        break;
    }
 
 
 
    // run shared js-env code - post-init
    (function () {
        return;
    }());
    switch (local.modeJs) {
 
 
 
    // run browser js-env code - post-init
    case 'browser':
        local.testCase_browser_nullCase = local.testCase_browser_nullCase || function (
            options,
            onError
        ) {
        /*
         * this function will test browsers's null-case handling-behavior-behavior
         */
            onError(null, options);
        };
 
        // run tests
        local.nop(local.modeTest &&
            document.querySelector('#testRunButton1') &&
            document.querySelector('#testRunButton1').click());
        break;
 
 
 
    // run node js-env code - post-init
    /* istanbul ignore next */
    case 'node':
        local.testCase_buildApidoc_default = local.testCase_buildApidoc_default || function (
            options,
            onError
        ) {
        /*
         * this function will test buildApidoc's default handling-behavior-behavior
         */
            options = { modulePathList: module.paths };
            local.buildApidoc(options, onError);
        };
 
        local.testCase_buildApp_default = local.testCase_buildApp_default || function (
            options,
            onError
        ) {
        /*
         * this function will test buildApp's default handling-behavior-behavior
         */
            local.testCase_buildReadme_default(options, local.onErrorThrow);
            local.testCase_buildLib_default(options, local.onErrorThrow);
            local.testCase_buildTest_default(options, local.onErrorThrow);
            local.testCase_buildCustomOrg_default(options, local.onErrorThrow);
            options = [];
            local.buildApp(options, onError);
        };
 
        local.testCase_buildCustomOrg_default = local.testCase_buildCustomOrg_default ||
            function (options, onError) {
            /*
             * this function will test buildCustomOrg's default handling-behavior
             */
                options = {};
                local.buildCustomOrg(options, onError);
            };
 
        local.testCase_buildLib_default = local.testCase_buildLib_default || function (
            options,
            onError
        ) {
        /*
         * this function will test buildLib's default handling-behavior
         */
            options = {};
            local.buildLib(options, onError);
        };
 
        local.testCase_buildReadme_default = local.testCase_buildReadme_default || function (
            options,
            onError
        ) {
        /*
         * this function will test buildReadme's default handling-behavior-behavior
         */
            options = {};
            local.buildReadme(options, onError);
        };
 
        local.testCase_buildTest_default = local.testCase_buildTest_default || function (
            options,
            onError
        ) {
        /*
         * this function will test buildTest's default handling-behavior
         */
            options = {};
            local.buildTest(options, onError);
        };
 
        local.testCase_webpage_default = local.testCase_webpage_default || function (
            options,
            onError
        ) {
        /*
         * this function will test webpage's default handling-behavior
         */
            options = { modeCoverageMerge: true, url: local.serverLocalHost + '?modeTest=1' };
            local.browserTest(options, onError);
        };
 
        // run test-server
        local.testRunServer(local);
        break;
    }
}());
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
1
1
 
1
1
1
 
1
1
6
6
6
6
5
 
 
 
 
 
 
var fs = require('fs');
var consoleService = require('./lib/consoleService');
 
module.exports.createMasterConsole = consoleService.createMasterConsole;
module.exports.createMonitorConsole = consoleService.createMonitorConsole;
module.exports.adminClient = require('./lib/client/client');
 
exports.modules = {};
fs.readdirSync(__dirname + '/lib/modules').forEach(function(filename) {
	Eif (/\.js$/.test(filename)) {
		var name = filename.substr(0, filename.lastIndexOf('.'));
		var _module = require('./lib/modules/' + name);
		if (!_module.moduleError) {
			exports.modules.__defineGetter__(name, function() {
				return _module;
			});
		}
	}
});
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
1
1
1
1
1
1
1
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'ConsoleService');
var MonitorAgent = require('./monitor/monitorAgent');
var EventEmitter = require('events').EventEmitter;
var MasterAgent = require('./master/masterAgent');
var schedule = require('pomelo-scheduler');
var protocol = require('./util/protocol');
var utils = require('./util/utils');
var util = require('util');
 
var MS_OF_SECOND = 1000;
 
/**
 * ConsoleService Constructor
 *
 * @class ConsoleService
 * @constructor
 * @param {Object} opts construct parameter
 *                 opts.type 	{String} server type, 'master', 'connector', etc.
 *                 opts.id 		{String} server id
 *                 opts.host 	{String} (monitor only) master server host
 *                 opts.port 	{String | Number} listen port for master or master port for monitor
 *                 opts.master  {Boolean} current service is master or monitor
 *                 opts.info 	{Object} more server info for current server, {id, serverType, host, port}
 * @api public
 */
var ConsoleService = function(opts) {
	EventEmitter.call(this);
	this.port = opts.port;
	this.env = opts.env;
	this.values = {};
	this.master = opts.master;
 
	this.modules = {};
	this.commands = {
		'list': listCommand,
		'enable': enableCommand,
		'disable': disableCommand
	};
 
	if (this.master) {
		this.authUser = opts.authUser || utils.defaultAuthUser;
		this.authServer = opts.authServer || utils.defaultAuthServerMaster;
		this.agent = new MasterAgent(this, opts);
	} else {
		this.type = opts.type;
		this.id = opts.id;
		this.host = opts.host;
		this.authServer = opts.authServer || utils.defaultAuthServerMonitor;
		this.agent = new MonitorAgent({
			consoleService: this,
			id: this.id,
			type: this.type,
			info: opts.info
		});
	}
};
 
util.inherits(ConsoleService, EventEmitter);
 
/**
 * start master or monitor
 *
 * @param {Function} cb callback function
 * @api public
 */
ConsoleService.prototype.start = function(cb) {
	if (this.master) {
		var self = this;
		this.agent.listen(this.port, function(err) {
			if (!!err) {
				utils.invokeCallback(cb, err);
				return;
			}
 
			exportEvent(self, self.agent, 'register');
			exportEvent(self, self.agent, 'disconnect');
			exportEvent(self, self.agent, 'reconnect');
			process.nextTick(function() {
				utils.invokeCallback(cb);
			});
		});
	} else {
		logger.info('try to connect master: %j, %j, %j', this.type, this.host, this.port);
		this.agent.connect(this.port, this.host, cb);
		exportEvent(this, this.agent, 'close');
	}
 
	exportEvent(this, this.agent, 'error');
 
	for (var mid in this.modules) {
		this.enable(mid);
	}
};
 
/**
 * stop console modules and stop master server
 *
 * @api public
 */
ConsoleService.prototype.stop = function() {
	for (var mid in this.modules) {
		this.disable(mid);
	}
	this.agent.close();
};
 
/**
 * register a new adminConsole module
 *
 * @param {String} moduleId adminConsole id/name
 * @param {Object} module module object
 * @api public
 */
ConsoleService.prototype.register = function(moduleId, module) {
	this.modules[moduleId] = registerRecord(this, moduleId, module);
};
 
/**
 * enable adminConsole module
 *
 * @param {String} moduleId adminConsole id/name
 * @api public
 */
ConsoleService.prototype.enable = function(moduleId) {
	var record = this.modules[moduleId];
	if (record && !record.enable) {
		record.enable = true;
		addToSchedule(this, record);
		return true;
	}
	return false;
};
 
/**
 * disable adminConsole module
 *
 * @param {String} moduleId adminConsole id/name
 * @api public
 */
ConsoleService.prototype.disable = function(moduleId) {
	var record = this.modules[moduleId];
	if (record && record.enable) {
		record.enable = false;
		if (record.schedule && record.jobId) {
			schedule.cancelJob(record.jobId);
			schedule.jobId = null;
		}
		return true;
	}
	return false;
};
 
/**
 * call concrete module and handler(monitorHandler,masterHandler,clientHandler)
 *
 * @param {String} moduleId adminConsole id/name
 * @param {String} method handler
 * @param {Object} msg message
 * @param {Function} cb callback function
 * @api public
 */
ConsoleService.prototype.execute = function(moduleId, method, msg, cb) {
	var self = this;
	var m = this.modules[moduleId];
	if (!m) {
		logger.error('unknown module: %j.', moduleId);
		cb('unknown moduleId:' + moduleId);
		return;
	}
 
	if (!m.enable) {
		logger.error('module %j is disable.', moduleId);
		cb('module ' + moduleId + ' is disable');
		return;
	}
 
	var module = m.module;
	if (!module || typeof module[method] !== 'function') {
		logger.error('module %j dose not have a method called %j.', moduleId, method);
		cb('module ' + moduleId + ' dose not have a method called ' + method);
		return;
	}
 
	var log = {
		action: 'execute',
		moduleId: moduleId,
		method: method,
		msg: msg
	}
 
	var aclMsg = aclControl(self.agent, 'execute', method, moduleId, msg);
	if (aclMsg !== 0 && aclMsg !== 1) {
		log['error'] = aclMsg;
		self.emit('admin-log', log, aclMsg);
		cb(new Error(aclMsg), null);
		return;
	}
 
	if (method === 'clientHandler') {
		self.emit('admin-log', log);
	}
 
	module[method](this.agent, msg, cb);
};
 
ConsoleService.prototype.command = function(command, moduleId, msg, cb) {
	var self = this;
	var fun = this.commands[command];
	if (!fun || typeof fun !== 'function') {
		cb('unknown command:' + command);
		return;
	}
 
	var log = {
		action: 'command',
		moduleId: moduleId,
		msg: msg
	}
 
	var aclMsg = aclControl(self.agent, 'command', null, moduleId, msg);
	if (aclMsg !== 0 && aclMsg !== 1) {
		log['error'] = aclMsg;
		self.emit('admin-log', log, aclMsg);
		cb(new Error(aclMsg), null);
		return;
	}
 
	self.emit('admin-log', log);
	fun(this, moduleId, msg, cb);
}
 
/**
 * set module data to a map
 *
 * @param {String} moduleId adminConsole id/name
 * @param {Object} value module data
 * @api public
 */
 
ConsoleService.prototype.set = function(moduleId, value) {
	this.values[moduleId] = value;
};
 
/**
 * get module data from map
 *
 * @param {String} moduleId adminConsole id/name
 * @api public
 */
ConsoleService.prototype.get = function(moduleId) {
	return this.values[moduleId];
};
 
/**
 * register a module service
 *
 * @param {Object} service consoleService object
 * @param {String} moduleId adminConsole id/name
 * @param {Object} module module object
 * @api private
 */
var registerRecord = function(service, moduleId, module) {
	var record = {
		moduleId: moduleId,
		module: module,
		enable: false
	};
 
	if (module.type && module.interval) {
		if (!service.master && record.module.type === 'push' || service.master && record.module.type !== 'push') {
			// push for monitor or pull for master(default)
			record.delay = module.delay || 0;
			record.interval = module.interval || 1;
			// normalize the arguments
			if (record.delay < 0) {
				record.delay = 0;
			}
			if (record.interval < 0) {
				record.interval = 1;
			}
			record.interval = Math.ceil(record.interval);
			record.delay *= MS_OF_SECOND;
			record.interval *= MS_OF_SECOND;
			record.schedule = true;
		}
	}
 
	return record;
};
 
/**
 * schedule console module
 *
 * @param {Object} service consoleService object
 * @param {Object} record  module object
 * @api private
 */
var addToSchedule = function(service, record) {
	if (record && record.schedule) {
		record.jobId = schedule.scheduleJob({
				start: Date.now() + record.delay,
				period: record.interval
			},
			doScheduleJob, {
				service: service,
				record: record
			});
	}
};
 
/**
 * run schedule job
 *
 * @param {Object} args argments
 * @api private
 */
var doScheduleJob = function(args) {
	var service = args.service;
	var record = args.record;
	if (!service || !record || !record.module || !record.enable) {
		return;
	}
 
	if (service.master) {
		record.module.masterHandler(service.agent, null, function(err) {
			logger.error('interval push should not have a callback.');
		});
	} else {
		record.module.monitorHandler(service.agent, null, function(err) {
			logger.error('interval push should not have a callback.');
		});
	}
};
 
/**
 * export closure function out
 *
 * @param {Function} outer outer function
 * @param {Function} inner inner function
 * @param {object} event
 * @api private
 */
var exportEvent = function(outer, inner, event) {
	inner.on(event, function() {
		var args = Array.prototype.slice.call(arguments, 0);
		args.unshift(event);
		outer.emit.apply(outer, args);
	});
};
 
/**
 * List current modules
 */
var listCommand = function(consoleService, moduleId, msg, cb) {
	var modules = consoleService.modules;
 
	var result = [];
	for (var moduleId in modules) {
		if (/^__\w+__$/.test(moduleId)) {
			continue;
		}
 
		result.push(moduleId);
	}
 
	cb(null, {
		modules: result
	});
};
 
/**
 * enable module in current server
 */
var enableCommand = function(consoleService, moduleId, msg, cb) {
	if (!moduleId) {
		logger.error('fail to enable admin module for ' + moduleId);
		cb('empty moduleId');
		return;
	}
 
	var modules = consoleService.modules;
	if (!modules[moduleId]) {
		cb(null, protocol.PRO_FAIL);
		return;
	}
 
	if (consoleService.master) {
		consoleService.enable(moduleId);
		consoleService.agent.notifyCommand("enable", moduleId, msg);
		cb(null, protocol.PRO_OK);
	} else {
		consoleService.enable(moduleId);
		cb(null, protocol.PRO_OK);
	}
};
 
/**
 * disable module in current server
 */
var disableCommand = function(consoleService, moduleId, msg, cb) {
	if (!moduleId) {
		logger.error('fail to enable admin module for ' + moduleId);
		cb('empty moduleId');
		return;
	}
 
	var modules = consoleService.modules;
	if (!modules[moduleId]) {
		cb(null, protocol.PRO_FAIL);
		return;
	}
 
	if (consoleService.master) {
		consoleService.disable(moduleId);
		consoleService.agent.notifyCommand("disable", moduleId, msg);
		cb(null, protocol.PRO_OK);
	} else {
		consoleService.disable(moduleId);
		cb(null, protocol.PRO_OK);
	}
};
 
var aclControl = function(agent, action, method, moduleId, msg) {
	if (action === 'execute') {
		if (method !== 'clientHandler' || moduleId !== '__console__') {
			return 0;
		}
 
		var signal = msg.signal;
		if (!signal || !(signal === 'stop' || signal === 'add' || signal === 'kill')) {
			return 0;
		}
	}
 
	var clientId = msg.clientId;
	if (!clientId) {
		return 'Unknow clientId';
	}
 
	var _client = agent.getClientById(clientId);
	if (_client && _client.info && _client.info.level) {
		var level = _client.info.level;
		if (level > 1) {
			return 'Command permission denied';
		}
	} else {
		return 'Client info error';
	}
	return 1;
}
 
/**
 * Create master ConsoleService
 *
 * @param {Object} opts construct parameter
 *                      opts.port {String | Number} listen port for master console
 */
module.exports.createMasterConsole = function(opts) {
	opts = opts || {};
	opts.master = true;
	return new ConsoleService(opts);
};
 
/**
 * Create monitor ConsoleService
 *
 * @param {Object} opts construct parameter
 *                      opts.type {String} server type, 'master', 'connector', etc.
 *                      opts.id {String} server id
 *                      opts.host {String} master server host
 *                      opts.port {String | Number} master port
 */
module.exports.createMonitorConsole = function(opts) {
	return new ConsoleService(opts);
};
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
 
 
 
 
 
1
1
 
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
1
1
1
 
1
 
/*!
 * Pomelo -- commandLine Client
 * Copyright(c) 2015 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
 
var MqttClient = require('../protocol/mqtt/mqttClient');
var protocol = require('../util/protocol');
// var io = require('socket.io-client');
var utils = require('../util/utils');
 
var Client = function(opt) {
	this.id = "";
	this.reqId = 1;
	this.callbacks = {};
	this.listeners = {};
	this.state = Client.ST_INITED;
	this.socket = null;
	opt = opt || {};
	this.username = opt['username'] || "";
	this.password = opt['password'] || "";
	this.md5 = opt['md5'] || false;
};
 
Client.prototype = {
	connect: function(id, host, port, cb) {
		this.id = id;
		var self = this;
 
		console.log('try to connect ' + host + ':' + port);
		this.socket = new MqttClient({
			id: id
		});
 
		this.socket.connect(host, port);
 
		// this.socket = io.connect('http://' + host + ':' + port, {
		// 	'force new connection': true,
		// 	'reconnect': false
		// });
 
		this.socket.on('connect', function() {
			self.state = Client.ST_CONNECTED;
			if (self.md5) {
				self.password = utils.md5(self.password);
			}
			self.doSend('register', {
				type: "client",
				id: id,
				username: self.username,
				password: self.password,
				md5: self.md5
			});
		});
 
		this.socket.on('register', function(res) {
			if (res.code !== protocol.PRO_OK) {
				cb(res.msg);
				return;
			}
 
			self.state = Client.ST_REGISTERED;
			cb();
		});
 
		this.socket.on('client', function(msg) {
			msg = protocol.parse(msg);
			if (msg.respId) {
				// response for request
				var cb = self.callbacks[msg.respId];
				delete self.callbacks[msg.respId];
				if (cb && typeof cb === 'function') {
					cb(msg.error, msg.body);
				}
			} else if (msg.moduleId) {
				// notify
				self.emit(msg.moduleId, msg);
			}
		});
 
		this.socket.on('error', function(err) {
			if (self.state < Client.ST_CONNECTED) {
				cb(err);
			}
 
			self.emit('error', err);
		});
 
		this.socket.on('disconnect', function(reason) {
			this.state = Client.ST_CLOSED;
			self.emit('close');
		});
	},
 
	request: function(moduleId, msg, cb) {
		var id = this.reqId++;
		// something dirty: attach current client id into msg
		msg = msg || {};
		msg.clientId = this.id;
		msg.username = this.username;
		var req = protocol.composeRequest(id, moduleId, msg);
		this.callbacks[id] = cb;
		this.doSend('client', req);
		// this.socket.emit('client', req);
	},
 
	notify: function(moduleId, msg) {
		// something dirty: attach current client id into msg
		msg = msg || {};
		msg.clientId = this.id;
		msg.username = this.username;
		var req = protocol.composeRequest(null, moduleId, msg);
		this.doSend('client', req);
		// this.socket.emit('client', req);
	},
 
	command: function(command, moduleId, msg, cb) {
		var id = this.reqId++;
		msg = msg || {};
		msg.clientId = this.id;
		msg.username = this.username;
		var commandReq = protocol.composeCommand(id, command, moduleId, msg);
		this.callbacks[id] = cb;
		this.doSend('client', commandReq);
		// this.socket.emit('client', commandReq);
	},
 
	doSend: function(topic, msg) {
		this.socket.send(topic, msg);
	},
 
	on: function(event, listener) {
		this.listeners[event] = this.listeners[event] || [];
		this.listeners[event].push(listener);
	},
 
	emit: function(event) {
		var listeners = this.listeners[event];
		if (!listeners || !listeners.length) {
			return;
		}
 
		var args = Array.prototype.slice.call(arguments, 1);
		var listener;
		for (var i = 0, l = listeners.length; i < l; i++) {
			listener = listeners[i];
			if (typeof listener === 'function') {
				listener.apply(null, args);
			}
		}
	}
};
 
Client.ST_INITED = 1;
Client.ST_CONNECTED = 2;
Client.ST_REGISTERED = 3;
Client.ST_CLOSED = 4;
 
module.exports = Client;
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
1
1
1
1
1
1
1
 
1
1
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
1
 
1
 
1
 
1
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'MasterAgent');
var MqttServer = require('../protocol/mqtt/mqttServer');
var EventEmitter = require('events').EventEmitter;
var MasterSocket = require('./masterSocket');
var protocol = require('../util/protocol');
var utils = require('../util/utils');
var Util = require('util');
 
var ST_INITED = 1;
var ST_STARTED = 2;
var ST_CLOSED = 3;
 
/**
 * MasterAgent Constructor
 *
 * @class MasterAgent
 * @constructor
 * @param {Object} opts construct parameter
 *                 opts.consoleService {Object} consoleService
 *                 opts.id             {String} server id
 *                 opts.type           {String} server type, 'master', 'connector', etc.
 *                 opts.socket         {Object} socket-io object
 *                 opts.reqId          {Number} reqId add by 1
 *                 opts.callbacks      {Object} callbacks
 *                 opts.state          {Number} MasterAgent state
 * @api public
 */
var MasterAgent = function(consoleService, opts) {
  EventEmitter.call(this);
  this.reqId = 1;
  this.idMap = {};
  this.msgMap = {};
  this.typeMap = {};
  this.clients = {};
  this.sockets = {};
  this.slaveMap = {};
  this.server = null;
  this.callbacks = {};
  this.state = ST_INITED;
  this.whitelist = opts.whitelist;
  this.consoleService = consoleService;
};
 
Util.inherits(MasterAgent, EventEmitter);
 
/**
 * master listen to a port and handle register and request
 *
 * @param {String} port
 * @api public
 */
MasterAgent.prototype.listen = function(port, cb) {
  if (this.state > ST_INITED) {
    logger.error('master agent has started or closed.');
    return;
  }
 
  this.state = ST_STARTED;
  this.server = new MqttServer();
  this.server.listen(port);
  // this.server = sio.listen(port);
  // this.server.set('log level', 0);
 
  cb = cb || function() {}
 
  var self = this;
  this.server.on('error', function(err) {
    self.emit('error', err);
    cb(err);
  });
 
  this.server.once('listening', function() {
    setImmediate(function() {
      cb();
    });
  });
 
  this.server.on('connection', function(socket) {
    // var id, type, info, registered, username;
    var masterSocket = new MasterSocket();
    masterSocket['agent'] = self;
    masterSocket['socket'] = socket;
 
    self.sockets[socket.id] = socket;
 
    socket.on('register', function(msg) {
      // register a new connection
      masterSocket.onRegister(msg);
    }); // end of on 'register'
 
    // message from monitor
    socket.on('monitor', function(msg) {
      masterSocket.onMonitor(msg);
    }); // end of on 'monitor'
 
    // message from client
    socket.on('client', function(msg) {
      masterSocket.onClient(msg);
    }); // end of on 'client'
 
    socket.on('reconnect', function(msg) {
      masterSocket.onReconnect(msg);
    });
 
    socket.on('disconnect', function() {
      masterSocket.onDisconnect();
    });
 
    socket.on('close', function() {
      masterSocket.onDisconnect();
    });
 
    socket.on('error', function(err) {
      masterSocket.onError(err);
    });
  }); // end of on 'connection'
}; // end of listen
 
/**
 * close master agent
 *
 * @api public
 */
MasterAgent.prototype.close = function() {
  if (this.state > ST_STARTED) {
    return;
  }
  this.state = ST_CLOSED;
  this.server.close();
};
 
/**
 * set module
 *
 * @param {String} moduleId module id/name
 * @param {Object} value module object
 * @api public
 */
MasterAgent.prototype.set = function(moduleId, value) {
  this.consoleService.set(moduleId, value);
};
 
/**
 * get module
 *
 * @param {String} moduleId module id/name
 * @api public
 */
MasterAgent.prototype.get = function(moduleId) {
  return this.consoleService.get(moduleId);
};
 
/**
 * getClientById
 *
 * @param {String} clientId
 * @api public
 */
MasterAgent.prototype.getClientById = function(clientId) {
  return this.clients[clientId];
};
 
/**
 * request monitor{master node} data from monitor
 *
 * @param {String} serverId
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @param {Function} callback function
 * @api public
 */
MasterAgent.prototype.request = function(serverId, moduleId, msg, cb) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  cb = cb || function() {}
 
  var curId = this.reqId++;
  this.callbacks[curId] = cb;
 
  if (!this.msgMap[serverId]) {
    this.msgMap[serverId] = {};
  }
 
  this.msgMap[serverId][curId] = {
    moduleId: moduleId,
    msg: msg
  }
 
  var record = this.idMap[serverId];
  if (!record) {
    cb(new Error('unknown server id:' + serverId));
    return false;
  }
 
  sendToMonitor(record.socket, curId, moduleId, msg);
 
  return true;
};
 
/**
 * request server data from monitor by serverInfo{host:port}
 *
 * @param {String} serverId
 * @param {Object} serverInfo
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @param {Function} callback function
 * @api public
 */
MasterAgent.prototype.requestServer = function(serverId, serverInfo, moduleId, msg, cb) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var record = this.idMap[serverId];
  if (!record) {
    utils.invokeCallback(cb, new Error('unknown server id:' + serverId));
    return false;
  }
 
  var curId = this.reqId++;
  this.callbacks[curId] = cb;
 
  if (utils.compareServer(record, serverInfo)) {
    sendToMonitor(record.socket, curId, moduleId, msg);
  } else {
    var slaves = this.slaveMap[serverId];
    for (var i = 0, l = slaves.length; i < l; i++) {
      if (utils.compareServer(slaves[i], serverInfo)) {
        sendToMonitor(slaves[i].socket, curId, moduleId, msg);
        break;
      }
    }
  }
 
  return true;
};
 
/**
 * notify a monitor{master node} by id without callback
 *
 * @param {String} serverId
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifyById = function(serverId, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var record = this.idMap[serverId];
  if (!record) {
    logger.error('fail to notifyById for unknown server id:' + serverId);
    return false;
  }
 
  sendToMonitor(record.socket, null, moduleId, msg);
 
  return true;
};
 
/**
 * notify a monitor by server{host:port} without callback
 *
 * @param {String} serverId
 * @param {Object} serverInfo{host:port}
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifyByServer = function(serverId, serverInfo, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var record = this.idMap[serverId];
  if (!record) {
    logger.error('fail to notifyByServer for unknown server id:' + serverId);
    return false;
  }
 
  if (utils.compareServer(record, serverInfo)) {
    sendToMonitor(record.socket, null, moduleId, msg);
  } else {
    var slaves = this.slaveMap[serverId];
    for (var i = 0, l = slaves.length; i < l; i++) {
      if (utils.compareServer(slaves[i], serverInfo)) {
        sendToMonitor(slaves[i].socket, null, moduleId, msg);
        break;
      }
    }
  }
  return true;
};
 
/**
 * notify slaves by id without callback
 *
 * @param {String} serverId
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifySlavesById = function(serverId, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var slaves = this.slaveMap[serverId];
  if (!slaves || slaves.length === 0) {
    logger.error('fail to notifySlavesById for unknown server id:' + serverId);
    return false;
  }
 
  broadcastMonitors(slaves, moduleId, msg);
  return true;
};
 
/**
 * notify monitors by type without callback
 *
 * @param {String} type serverType
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifyByType = function(type, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var list = this.typeMap[type];
  if (!list || list.length === 0) {
    logger.error('fail to notifyByType for unknown server type:' + type);
    return false;
  }
  broadcastMonitors(list, moduleId, msg);
  return true;
};
 
/**
 * notify all the monitors without callback
 *
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifyAll = function(moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
  broadcastMonitors(this.idMap, moduleId, msg);
  return true;
};
 
/**
 * notify a client by id without callback
 *
 * @param {String} clientId
 * @param {String} moduleId module id/name
 * @param {Object} msg
 * @api public
 */
MasterAgent.prototype.notifyClient = function(clientId, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
 
  var record = this.clients[clientId];
  if (!record) {
    logger.error('fail to notifyClient for unknown client id:' + clientId);
    return false;
  }
  sendToClient(record.socket, null, moduleId, msg);
};
 
MasterAgent.prototype.notifyCommand = function(command, moduleId, msg) {
  if (this.state > ST_STARTED) {
    return false;
  }
  broadcastCommand(this.idMap, command, moduleId, msg);
  return true;
};
 
/**
 * add monitor,client to connection -- idMap
 *
 * @param {Object} agent agent object
 * @param {String} id
 * @param {String} type serverType
 * @param {Object} socket socket-io object
 * @api private
 */
var addConnection = function(agent, id, type, pid, info, socket) {
  var record = {
    id: id,
    type: type,
    pid: pid,
    info: info,
    socket: socket
  };
  if (type === 'client') {
    agent.clients[id] = record;
  } else {
    if (!agent.idMap[id]) {
      agent.idMap[id] = record;
      var list = agent.typeMap[type] = agent.typeMap[type] || [];
      list.push(record);
    } else {
      var slaves = agent.slaveMap[id] = agent.slaveMap[id] || [];
      slaves.push(record);
    }
  }
  return record;
};
 
/**
 * remove monitor,client connection -- idMap
 *
 * @param {Object} agent agent object
 * @param {String} id
 * @param {String} type serverType
 * @api private
 */
var removeConnection = function(agent, id, type, info) {
  if (type === 'client') {
    delete agent.clients[id];
  } else {
    // remove master node in idMap and typeMap
    var record = agent.idMap[id];
    if (!record) {
      return;
    }
    var _info = record['info']; // info {host, port}
    if (utils.compareServer(_info, info)) {
      delete agent.idMap[id];
      var list = agent.typeMap[type];
      if (list) {
        for (var i = 0, l = list.length; i < l; i++) {
          if (list[i].id === id) {
            list.splice(i, 1);
            break;
          }
        }
        if (list.length === 0) {
          delete agent.typeMap[type];
        }
      }
    } else {
      // remove slave node in slaveMap
      var slaves = agent.slaveMap[id];
      if (slaves) {
        for (var i = 0, l = slaves.length; i < l; i++) {
          if (utils.compareServer(slaves[i]['info'], info)) {
            slaves.splice(i, 1);
            break;
          }
        }
        if (slaves.length === 0) {
          delete agent.slaveMap[id];
        }
      }
    }
  }
};
 
/**
 * send msg to monitor
 *
 * @param {Object} socket socket-io object
 * @param {Number} reqId request id
 * @param {String} moduleId module id/name
 * @param {Object} msg message
 * @api private
 */
var sendToMonitor = function(socket, reqId, moduleId, msg) {
  doSend(socket, 'monitor', protocol.composeRequest(reqId, moduleId, msg));
};
 
/**
 * send msg to client
 *
 * @param {Object} socket socket-io object
 * @param {Number} reqId request id
 * @param {String} moduleId module id/name
 * @param {Object} msg message
 * @api private
 */
var sendToClient = function(socket, reqId, moduleId, msg) {
  doSend(socket, 'client', protocol.composeRequest(reqId, moduleId, msg));
};
 
var doSend = function(socket, topic, msg) {
  socket.send(topic, msg);
}
 
/**
 * broadcast msg to monitor
 *
 * @param {Object} record registered modules
 * @param {String} moduleId module id/name
 * @param {Object} msg message
 * @api private
 */
var broadcastMonitors = function(records, moduleId, msg) {
  msg = protocol.composeRequest(null, moduleId, msg);
 
  if (records instanceof Array) {
    for (var i = 0, l = records.length; i < l; i++) {
      var socket = records[i].socket;
      doSend(socket, 'monitor', msg);
    }
  } else {
    for (var id in records) {
      var socket = records[id].socket;
      doSend(socket, 'monitor', msg);
    }
  }
};
 
var broadcastCommand = function(records, command, moduleId, msg) {
  msg = protocol.composeCommand(null, command, moduleId, msg);
 
  if (records instanceof Array) {
    for (var i = 0, l = records.length; i < l; i++) {
      var socket = records[i].socket;
      doSend(socket, 'monitor', msg);
    }
  } else {
    for (var id in records) {
      var socket = records[id].socket;
      doSend(socket, 'monitor', msg);
    }
  }
};
 
MasterAgent.prototype.doAuthUser = function(msg, socket, cb) {
  if (!msg.id) {
    // client should has a client id
    return cb(new Error('client should has a client id'));
  }
 
  var self = this;
  var username = msg.username;
  if (!username) {
    // client should auth with username
    doSend(socket, 'register', {
      code: protocol.PRO_FAIL,
      msg: 'client should auth with username'
    });
    return cb(new Error('client should auth with username'));
  }
 
  var authUser = self.consoleService.authUser;
  var env = self.consoleService.env;
  authUser(msg, env, function(user) {
    if (!user) {
      // client should auth with username
      doSend(socket, 'register', {
        code: protocol.PRO_FAIL,
        msg: 'client auth failed with username or password error'
      });
      return cb(new Error('client auth failed with username or password error'));
    }
 
    if (self.clients[msg.id]) {
      doSend(socket, 'register', {
        code: protocol.PRO_FAIL,
        msg: 'id has been registered. id:' + msg.id
      });
      return cb(new Error('id has been registered. id:' + msg.id));
    }
 
    logger.info('client user : ' + username + ' login to master');
    addConnection(self, msg.id, msg.type, null, user, socket);
    doSend(socket, 'register', {
      code: protocol.PRO_OK,
      msg: 'ok'
    });
 
    cb();
  });
};
 
MasterAgent.prototype.doAuthServer = function(msg, socket, cb) {
  var self = this;
  var authServer = self.consoleService.authServer;
  var env = self.consoleService.env;
  authServer(msg, env, function(status) {
    if (status !== 'ok') {
      doSend(socket, 'register', {
        code: protocol.PRO_FAIL,
        msg: 'server auth failed'
      });
      cb(new Error('server auth failed'));
      return;
    }
 
    var record = addConnection(self, msg.id, msg.serverType, msg.pid, msg.info, socket);
 
    doSend(socket, 'register', {
      code: protocol.PRO_OK,
      msg: 'ok'
    });
    msg.info = msg.info || {}
    msg.info.pid = msg.pid;
    self.emit('register', msg.info);
    cb(null);
  });
};
 
MasterAgent.prototype.doSend = doSend;
 
MasterAgent.prototype.sendToMonitor = sendToMonitor;
 
MasterAgent.prototype.addConnection = addConnection;
 
MasterAgent.prototype.removeConnection = removeConnection;
 
module.exports = MasterAgent;
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
1
1
1
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
1
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'MasterSocket');
var Constants = require('../util/constants');
var protocol = require('../util/protocol');
 
var MasterSocket = function() {
	this.id = null;
	this.type = null;
	this.info = null;
	this.agent = null;
	this.socket = null;
	this.username = null;
	this.registered = false;
}
 
MasterSocket.prototype.onRegister = function(msg) {
	if (!msg || !msg.type) {
		return;
	}
 
	var self = this;
	var serverId = msg.id;
	var serverType = msg.type;
	var socket = this.socket;
 
	if (serverType == Constants.TYPE_CLIENT) {
		// client connection not join the map
		this.id = serverId;
		this.type = serverType;
		this.info = 'client';
		this.agent.doAuthUser(msg, socket, function(err) {
			if (err) {
				return socket.disconnect();
			}
 
			self.username = msg.username;
			self.registered = true;
		});
		return;
	} // end of if(serverType === 'client')
 
	if (serverType == Constants.TYPE_MONITOR) {
		if (!serverId) {
			return;
		}
 
		// if is a normal server
		this.id = serverId;
		this.type = msg.serverType;
		this.info = msg.info;
		this.agent.doAuthServer(msg, socket, function(err) {
			if (err) {
				return socket.disconnect();
			}
 
			self.registered = true;
		});
 
		this.repushQosMessage(serverId);
		return;
	} // end of if(serverType === 'monitor') 
 
	this.agent.doSend(socket, 'register', {
		code: protocol.PRO_FAIL,
		msg: 'unknown auth master type'
	});
 
	socket.disconnect();
}
 
MasterSocket.prototype.onMonitor = function(msg) {
	var socket = this.socket;
	if (!this.registered) {
		// not register yet, ignore any message
		// kick connections
		socket.disconnect();
		return;
	}
 
	var self = this;
	var type = this.type;
	if (type === Constants.TYPE_CLIENT) {
		logger.error('invalid message from monitor, but current connect type is client.');
		return;
	}
 
	msg = protocol.parse(msg);
	var respId = msg.respId;
	if (respId) {
		// a response from monitor
		var cb = self.agent.callbacks[respId];
		if (!cb) {
			logger.warn('unknown resp id:' + respId);
			return;
		}
 
		var id = this.id;
		if (self.agent.msgMap[id]) {
			delete self.agent.msgMap[id][respId];
		}
		delete self.agent.callbacks[respId];
		return cb(msg.error, msg.body);
	}
 
	// a request or a notify from monitor
	self.agent.consoleService.execute(msg.moduleId, 'masterHandler', msg.body, function(err, res) {
		if (protocol.isRequest(msg)) {
			var resp = protocol.composeResponse(msg, err, res);
			if (resp) {
				self.agent.doSend(socket, 'monitor', resp);
			}
		} else {
			//notify should not have a callback
			logger.warn('notify should not have a callback.');
		}
	});
}
 
MasterSocket.prototype.onClient = function(msg) {
	var socket = this.socket;
	if (!this.registered) {
		// not register yet, ignore any message
		// kick connections
		return socket.disconnect();
	}
 
	var type = this.type;
	if (type !== Constants.TYPE_CLIENT) {
		logger.error('invalid message to client, but current connect type is ' + type);
		return;
	}
 
	msg = protocol.parse(msg);
 
	var msgCommand = msg.command;
	var msgModuleId = msg.moduleId;
	var msgBody = msg.body;
 
	var self = this;
 
	if (msgCommand) {
		// a command from client
		self.agent.consoleService.command(msgCommand, msgModuleId, msgBody, function(err, res) {
			if (protocol.isRequest(msg)) {
				var resp = protocol.composeResponse(msg, err, res);
				if (resp) {
					self.agent.doSend(socket, 'client', resp);
				}
			} else {
				//notify should not have a callback
				logger.warn('notify should not have a callback.');
			}
		});
	} else {
		// a request or a notify from client
		// and client should not have any response to master for master would not request anything from client
		self.agent.consoleService.execute(msgModuleId, 'clientHandler', msgBody, function(err, res) {
			if (protocol.isRequest(msg)) {
				var resp = protocol.composeResponse(msg, err, res);
				if (resp) {
					self.agent.doSend(socket, 'client', resp);
				}
			} else {
				//notify should not have a callback
				logger.warn('notify should not have a callback.');
			}
		});
	}
}
 
MasterSocket.prototype.onReconnect = function(msg, pid) {
	// reconnect a new connection
	if (!msg || !msg.type) {
		return;
	}
 
	var serverId = msg.id;
	if (!serverId) {
		return;
	}
 
	var socket = this.socket;
 
	// if is a normal server
	if (this.agent.idMap[serverId]) {
		// id has been registered
		this.agent.doSend(socket, 'reconnect_ok', {
			code: protocol.PRO_FAIL,
			msg: 'id has been registered. id:' + serverId
		});
		return;
	}
 
	var msgServerType = msg.serverType;
	var record = this.agent.addConnection(this.agent, serverId, msgServerType, msg.pid, msg.info, socket);
 
	this.id = serverId;
	this.type = msgServerType;
	this.registered = true;
	msg.info.pid = pid;
	this.info = msg.info;
	this.agent.doSend(socket, 'reconnect_ok', {
		code: protocol.PRO_OK,
		msg: 'ok'
	});
 
	this.agent.emit('reconnect', msg.info);
 
	this.repushQosMessage(serverId);
}
 
MasterSocket.prototype.onDisconnect = function() {
	var socket = this.socket;
	if (socket) {
		delete this.agent.sockets[socket.id];
	}
 
	var registered = this.registered;
	if (!registered) {
		return;
	}
 
	var id = this.id;
	var type = this.type;
	var info = this.info;
	var username = this.username;
 
	logger.debug('disconnect %s %s %j', id, type, info);
	if (registered) {
		this.agent.removeConnection(this.agent, id, type, info);
		this.agent.emit('disconnect', id, type, info);
	}
 
	if (type === Constants.TYPE_CLIENT && registered) {
		logger.info('client user ' + username + ' exit');
	}
 
	this.registered = false;
	this.id = null;
	this.type = null;
}
 
MasterSocket.prototype.repushQosMessage = function(serverId) {
	var socket = this.socket;
	// repush qos message
	var qosMsgs = this.agent.msgMap[serverId];
 
	if (!qosMsgs) {
		return;
	}
 
	logger.debug('repush qos message %j', qosMsgs);
 
	for (var reqId in qosMsgs) {
		var qosMsg = qosMsgs[reqId];
		var moduleId = qosMsg['moduleId'];
		var tmsg = qosMsg['msg'];
 
		this.agent.sendToMonitor(socket, reqId, moduleId, tmsg);
	}
}
 
MasterSocket.prototype.onError = function(err) {
	// logger.error('server %s error %s', this.id, err.stack);
	// this.onDisconnect();
}
 
module.exports = MasterSocket;
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
 
 
 
 
1
1
1
 
1
 
1
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
/*!
 * Pomelo -- consoleModule monitorLog
 * Copyright(c) 2012 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
var exec = require('child_process').exec;
var path = require('path');
 
var DEFAULT_INTERVAL = 5 * 60;		// in second
 
module.exports = function(opts) {
	return new Module(opts);
};
 
module.exports.moduleId = 'monitorLog';
 
/**
 * Initialize a new 'Module' with the given 'opts'
 *
 * @class Module
 * @constructor
 * @param {object} opts
 * @api public
 */
var Module = function(opts) {
	opts = opts || {};
	this.root = opts.path;
	this.interval = opts.interval || DEFAULT_INTERVAL;
};
 
 /**
 * collect monitor data from monitor
 *
 * @param {Object} agent monitorAgent object
 * @param {Object} msg client message
 * @param {Function} cb callback function
 * @api public
 */
Module.prototype.monitorHandler = function(agent, msg, cb) {
	if(!msg.logfile) {
		cb(new Error('logfile should not be empty'));
		return;
	}
 
	var serverId = agent.id;
	fetchLogs(this.root, msg, function (data) {
		cb(null, {serverId: serverId, body: data});
	});
};
 
/**
 * Handle client request
 *
 * @param {Object} agent masterAgent object
 * @param {Object} msg client message
 * @param {Function} cb callback function
 * @api public
 */
Module.prototype.clientHandler = function(agent, msg, cb) {
	agent.request(msg.serverId, module.exports.moduleId, msg, function(err, res) {
		if(err) {
			logger.error('fail to run log for ' + err.stack);
			return;
		}
		cb(null, res);
	});
};
 
//get the latest logs
var fetchLogs = function(root, msg, callback) {
	var number = msg.number;
	var logfile = msg.logfile;
	var serverId = msg.serverId;
	var filePath = path.join(root, getLogFileName(logfile, serverId));
 
	var endLogs = [];
	exec('tail -n ' + number + ' ' + filePath, function(error, output) {
		var endOut = [];
		output = output.replace(/^\s+|\s+$/g, "").split(/\s+/);
 
		for(var i=5; i<output.length; i+=6) {
			endOut.push(output[i]);
		}
 
		var endLength=endOut.length;
		for(var j=0; j<endLength; j++) {
			var map = {};
			var json;
			try{
				json = JSON.parse(endOut[j]);
			} catch(e) {
				logger.error('the log cannot parsed to json, '+e);
				continue;
			}
			map.time = json.time;
			map.route = json.route || json.service;
			map.serverId = serverId;
			map.timeUsed = json.timeUsed;
			map.params = endOut[j];
			endLogs.push(map);
		}
 
		callback({logfile:logfile,dataArray:endLogs});
	});
};
 
var getLogFileName = function(logfile, serverId) {
	return logfile + '-' + serverId + '.log';
};
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
 
 
 
 
1
1
 
1
1
 
1
 
 
 
1
 
1
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
/*!
 * Pomelo -- consoleModule nodeInfo processInfo
 * Copyright(c) 2012 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
var monitor = require('pomelo-monitor');
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
 
var DEFAULT_INTERVAL = 5 * 60;		// in second
var DEFAULT_DELAY = 10;						// in second
 
module.exports = function(opts) {
	return new Module(opts);
};
 
module.exports.moduleId = 'nodeInfo';
 
var Module = function(opts) {
	opts = opts || {};
	this.type = opts.type || 'pull';
	this.interval = opts.interval || DEFAULT_INTERVAL;
	this.delay = opts.delay || DEFAULT_DELAY;
};
 
Module.prototype.monitorHandler = function(agent, msg, cb) {
	var serverId = agent.id;
	var pid = process.pid;
	var params = {
		serverId: serverId,
		pid: pid
	};
	monitor.psmonitor.getPsInfo(params, function (err, data) {
		agent.notify(module.exports.moduleId, {serverId: agent.id, body: data});
	});
 
};
 
Module.prototype.masterHandler = function(agent, msg, cb) {
	if(!msg) {
		agent.notifyAll(module.exports.moduleId);
		return;
	}
 
	var body=msg.body;
	var data = agent.get(module.exports.moduleId);
	if(!data) {
		data = {};
		agent.set(module.exports.moduleId, data);
	}
 
	data[msg.serverId] = body;
};
 
Module.prototype.clientHandler = function(agent, msg, cb) {
	cb(null, agent.get(module.exports.moduleId) || {});
};
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
1
1
 
1
1
1
 
 
 
1
1
 
1
 
 
 
 
 
 
 
1
1
 
 
1
 
1
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
var utils = require('../util/utils');
 
var profiler = null;
try {
	profiler = require('v8-profiler');
} catch(e) {
}
 
var fs = require('fs');
var ProfileProxy = require('../util/profileProxy');
 
module.exports = function(opts) {
	if (!profiler) {
		return {};
	} else {
		return new Module(opts);
	}
};
 
Eif (!profiler) {
	module.exports.moduleError = 1;
}
 
module.exports.moduleId = 'profiler';
 
var Module = function(opts) {
	if(opts && opts.isMaster) {
		this.proxy = new ProfileProxy();
	}
};
 
Module.prototype.monitorHandler = function(agent, msg, cb) {
	var type = msg.type, action = msg.action, uid = msg.uid, result = null;
	if (type === 'CPU') {
		if (action === 'start') {
			profiler.startProfiling();
		} else {
			result = profiler.stopProfiling();
			var res = {};
			res.head = result.getTopDownRoot();
			res.bottomUpHead = result.getBottomUpRoot();
			res.msg = msg;
			agent.notify(module.exports.moduleId, {clientId: msg.clientId, type: type, body: res});
		}
	} else {
		var snapshot = profiler.takeSnapshot();
    var appBase = path.dirname(require.main.filename);
		var name = appBase + '/logs/' + utils.format(new Date()) + '.log';
		var log = fs.createWriteStream(name, {'flags': 'a'});
		var data;
		snapshot.serialize({
			onData: function (chunk, size) {
				chunk = chunk + '';
				data = {
					method:'Profiler.addHeapSnapshotChunk',
					params:{
						uid: uid,
						chunk: chunk
					}
				};
				log.write(chunk);
				agent.notify(module.exports.moduleId, {clientId: msg.clientId, type: type, body: data});
			},
			onEnd: function () {
				agent.notify(module.exports.moduleId, {clientId: msg.clientId, type: type, body: {params: {uid: uid}}});
				profiler.deleteAllSnapshots();
			}
		});
	}
};
 
Module.prototype.masterHandler = function(agent, msg, cb) {
	if(msg.type === 'CPU') {
		this.proxy.stopCallBack(msg.body, msg.clientId, agent);
	} else {
		this.proxy.takeSnapCallBack(msg.body);
	}
};
 
Module.prototype.clientHandler = function(agent, msg, cb) {
	if(msg.action === 'list') {
		list(agent, msg, cb);
		return;
	}
 
	if(typeof msg === 'string') {
		msg = JSON.parse(msg);
	}
	var id = msg.id;
	var command = msg.method.split('.');
	var method = command[1];
	var params = msg.params;
	var clientId = msg.clientId;
 
	if (!this.proxy[method] || typeof this.proxy[method] !== 'function') {
		return;
	}
 
	this.proxy[method](id, params, clientId, agent);
};
 
var list = function(agent, msg, cb) {
	var servers = [];
	var idMap = agent.idMap;
 
	for(var sid in idMap){
		servers.push(sid);
	}
	cb(null, servers);
};
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
 
 
 
 
1
1
1
1
1
1
1
 
1
 
 
 
1
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
/*!
 * Pomelo -- consoleModule runScript
 * Copyright(c) 2012 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
var monitor = require('pomelo-monitor');
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
var monitor = require('pomelo-monitor');
var vm = require('vm');
var fs = require('fs');
var util = require('util');
var path = require('path');
 
module.exports = function(opts) {
    return new Module(opts);
};
 
module.exports.moduleId = "scripts";
 
var Module = function(opts) {
    this.app = opts.app;
    this.root = opts.path;
    this.commands = {
        'list': list,
        'get': get,
        'save': save,
        'run': run
    };
};
 
Module.prototype.monitorHandler = function(agent, msg, cb) {
    var context = {
        app: this.app,
        require: require,
        os: require('os'),
        fs: require('fs'),
        process: process,
        util: util
    };
    try {
        vm.runInNewContext(msg.script, context);
 
        var result = context.result;
        if (!result) {
            cb(null, "script result should be assigned to result value to script module context");
        } else {
            cb(null, result);
        }
    } catch (e) {
        cb(null, e.toString());
    }
 
    //cb(null, vm.runInContext(msg.script, context));
};
 
Module.prototype.clientHandler = function(agent, msg, cb) {
    var fun = this.commands[msg.command];
    if (!fun || typeof fun !== 'function') {
        cb('unknown command:' + msg.command);
        return;
    }
 
    fun(this, agent, msg, cb);
};
 
/**
 * List server id and scripts file name
 */
var list = function(scriptModule, agent, msg, cb) {
    var servers = [];
    var scripts = [];
    var idMap = agent.idMap;
 
    for (var sid in idMap) {
        servers.push(sid);
    }
 
    fs.readdir(scriptModule.root, function(err, filenames) {
        if (err) {
            filenames = [];
        }
        for (var i = 0, l = filenames.length; i < l; i++) {
            scripts.push(filenames[i]);
        }
 
        cb(null, {
            servers: servers,
            scripts: scripts
        });
    });
};
 
/**
 * Get the content of the script file
 */
var get = function(scriptModule, agent, msg, cb) {
    var filename = msg.filename;
    if (!filename) {
        cb('empty filename');
        return;
    }
 
    fs.readFile(path.join(scriptModule.root, filename), 'utf-8', function(err, data) {
        if (err) {
            logger.error('fail to read script file:' + filename + ', ' + err.stack);
            cb('fail to read script with name:' + filename);
        }
 
        cb(null, data);
    });
};
 
/**
 * Save a script file that posted from admin console
 */
var save = function(scriptModule, agent, msg, cb) {
    var filepath = path.join(scriptModule.root, msg.filename);
 
    fs.writeFile(filepath, msg.body, function(err) {
        if (err) {
            logger.error('fail to write script file:' + msg.filename + ', ' + err.stack);
            cb('fail to write script file:' + msg.filename);
            return;
        }
 
        cb();
    });
};
 
/**
 * Run the script on the specified server
 */
var run = function(scriptModule, agent, msg, cb) {
    agent.request(msg.serverId, module.exports.moduleId, msg, function(err, res) {
        if (err) {
            logger.error('fail to run script for ' + err.stack);
            return;
        }
        cb(null, res);
    });
};
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
 
 
 
 
1
1
 
1
1
 
1
 
 
 
1
 
1
 
 
 
 
 
 
1
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
/*!
 * Pomelo -- consoleModule systemInfo
 * Copyright(c) 2012 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
var monitor = require('pomelo-monitor');
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
 
var DEFAULT_INTERVAL = 5 * 60;		// in second
var DEFAULT_DELAY = 10;						// in second
 
module.exports = function(opts) {
	return new Module(opts);
};
 
module.exports.moduleId = 'systemInfo';
 
var Module = function(opts) {
	opts = opts || {};
	this.type = opts.type || 'pull';
	this.interval = opts.interval || DEFAULT_INTERVAL;
	this.delay = opts.delay || DEFAULT_DELAY;
};
 
Module.prototype.monitorHandler = function(agent, msg, cb) {
	//collect data
	monitor.sysmonitor.getSysInfo(function (err, data) {
		agent.notify(module.exports.moduleId, {serverId: agent.id, body: data});
	});
};
 
Module.prototype.masterHandler = function(agent, msg) {
	if(!msg) {
		agent.notifyAll(module.exports.moduleId);
		return;
	}
 
	var body = msg.body;
 
	var oneData = {
		Time:body.iostat.date,hostname:body.hostname,serverId:msg.serverId,cpu_user:body.iostat.cpu.cpu_user,
		cpu_nice:body.iostat.cpu.cpu_nice,cpu_system:body.iostat.cpu.cpu_system,cpu_iowait:body.iostat.cpu.cpu_iowait,
		cpu_steal:body.iostat.cpu.cpu_steal,cpu_idle:body.iostat.cpu.cpu_idle,tps:body.iostat.disk.tps,
		kb_read:body.iostat.disk.kb_read,kb_wrtn:body.iostat.disk.kb_wrtn,kb_read_per:body.iostat.disk.kb_read_per,
		kb_wrtn_per:body.iostat.disk.kb_wrtn_per,totalmem:body.totalmem,freemem:body.freemem,'free/total':(body.freemem/body.totalmem),
		m_1:body.loadavg[0],m_5:body.loadavg[1],m_15:body.loadavg[2]
	};
 
	var data = agent.get(module.exports.moduleId);
	if(!data) {
		data = {};
		agent.set(module.exports.moduleId, data);
	}
 
	data[msg.serverId] = oneData;
};
 
Module.prototype.clientHandler = function(agent, msg, cb) {
	cb(null, agent.get(module.exports.moduleId) || {});
};
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
 
 
 
 
 
1
1
1
1
1
1
1
 
1
 
 
 
1
 
1
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
/*!
 * Pomelo -- consoleModule watchServer
 * Copyright(c) 2013 fantasyni <fantasyni@163.com>
 * MIT Licensed
 */
var logger = require('pomelo-logger').getLogger('pomelo-admin', __filename);
var countDownLatch = require('../util/countDownLatch');
var monitor = require('pomelo-monitor');
var utils = require('../util/utils');
var util = require('util');
var fs = require('fs');
var vm = require('vm');
 
module.exports = function(opts) {
	return new Module(opts);
};
 
module.exports.moduleId = 'watchServer';
 
var Module = function(opts) {
	opts = opts || {};
	this.app = opts.app;
};
 
Module.prototype.monitorHandler = function(agent, msg, cb) {
	var comd = msg['comd'];
	var context = msg['context'];
	var param = msg['param'];
	var app = this.app;
 
	var handle = 'monitor';
 
	switch (comd) {
		case 'servers':
			showServers(handle, agent, comd, context, cb);
			break;
		case 'connections':
			showConnections(handle, agent, app, comd, context, cb);
			break;
		case 'logins':
			showLogins(handle, agent, app, comd, context, cb);
			break;
		case 'modules':
			showModules(handle, agent, comd, context, cb);
			break;
		case 'status':
			showStatus(handle, agent, comd, context, cb);
			break;
		case 'config':
			showConfig(handle, agent, app, comd, context, param, cb);
			break;
		case 'proxy':
			showProxy(handle, agent, app, comd, context, param, cb);
			break;
		case 'handler':
			showHandler(handle, agent, app, comd, context, param, cb);
			break;
		case 'components':
			showComponents(handle, agent, app, comd, context, param, cb);
			break;
		case 'settings':
			showSettings(handle, agent, app, comd, context, param, cb);
			break;
		case 'cpu':
			dumpCPU(handle, agent, comd, context, param, cb);
			break;
		case 'memory':
			dumpMemory(handle, agent, comd, context, param, cb);
			break;
		case 'get':
			getApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'set':
			setApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'enable':
			enableApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'disable':
			disableApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'run':
			runScript(handle, agent, app, comd, context, param, cb);
			break;
		default:
			showError(handle, agent, comd, context, cb);
	}
};
 
Module.prototype.clientHandler = function(agent, msg, cb) {
	var comd = msg['comd'];
	var context = msg['context'];
	var param = msg['param'];
	var app = this.app; // master app
 
	if (!comd || !context) {
		cb('lack of comd or context param');
		return;
	}
 
	var handle = 'client';
	switch (comd) {
		case 'servers':
			showServers(handle, agent, comd, context, cb);
			break;
		case 'connections':
			showConnections(handle, agent, app, comd, context, cb);
			break;
		case 'logins':
			showLogins(handle, agent, app, comd, context, cb);
			break;
		case 'modules':
			showModules(handle, agent, comd, context, cb);
			break;
		case 'status':
			showStatus(handle, agent, comd, context, cb);
			break;
		case 'config':
			showConfig(handle, agent, app, comd, context, param, cb);
			break;
		case 'proxy':
			showProxy(handle, agent, app, comd, context, param, cb);
			break;
		case 'handler':
			showHandler(handle, agent, app, comd, context, param, cb);
			break;
		case 'components':
			showComponents(handle, agent, app, comd, context, param, cb);
			break;
		case 'settings':
			showSettings(handle, agent, app, comd, context, param, cb);
			break;
		case 'cpu':
			dumpCPU(handle, agent, comd, context, param, cb);
			break;
		case 'memory':
			dumpMemory(handle, agent, comd, context, param, cb);
			break;
		case 'get':
			getApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'set':
			setApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'enable':
			enableApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'disable':
			disableApp(handle, agent, app, comd, context, param, cb);
			break;
		case 'run':
			runScript(handle, agent, app, comd, context, param, cb);
			break;
		default:
			showError(handle, agent, comd, context, cb);
	}
};
 
function showServers(handle, agent, comd, context, cb) {
	if (handle === 'client') {
		var sid, record;
		var serverInfo = {};
		var count = utils.size(agent.idMap);
		var latch = countDownLatch.createCountDownLatch(count, function() {
			cb(null, {
				msg: serverInfo
			});
		});
 
		for (sid in agent.idMap) {
			record = agent.idMap[sid];
			agent.request(record.id, module.exports.moduleId, {
				comd: comd,
				context: context
			}, function(msg) {
				serverInfo[msg.serverId] = msg.body;
				latch.done();
			});
		}
	} else if (handle === 'monitor') {
		var serverId = agent.id;
		var serverType = agent.type;
		var info = agent.info;
		var pid = process.pid;
		var heapUsed = (process.memoryUsage().heapUsed / (1000 * 1000)).toFixed(2);
		var uptime = (process.uptime() / 60).toFixed(2);
		cb({
			serverId: serverId,
			body: {
				serverId: serverId,
				serverType: serverType,
				host: info['host'],
				port: info['port'],
				pid: pid,
				heapUsed: heapUsed,
				uptime: uptime
			}
		});
	}
 
}
 
function showConnections(handle, agent, app, comd, context, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			var sid, record;
			var serverInfo = {};
			var count = 0;
			for (var key in agent.idMap) {
				if (agent.idMap[key].info.frontend === 'true') {
					count++;
				}
			}
			var latch = countDownLatch.createCountDownLatch(count, function() {
				cb(null, {
					msg: serverInfo
				});
			});
 
			for (sid in agent.idMap) {
				record = agent.idMap[sid];
				if (record.info.frontend === 'true') {
					agent.request(record.id, module.exports.moduleId, {
						comd: comd,
						context: context
					}, function(msg) {
						serverInfo[msg.serverId] = msg.body;
						latch.done();
					});
				}
			}
		} else {
			var record = agent.idMap[context];
			if (!record) {
				cb("the server " + context + " not exist");
			}
			if (record.info.frontend === 'true') {
				agent.request(record.id, module.exports.moduleId, {
					comd: comd,
					context: context
				}, function(msg) {
					var serverInfo = {};
					serverInfo[msg.serverId] = msg.body;
					cb(null, {
						msg: serverInfo
					});
				});
			} else {
				cb('\nthis command should be applied to frontend server\n');
			}
		}
	} else if (handle === 'monitor') {
		var connection = app.components.__connection__;
		if (!connection) {
			cb({
				serverId: agent.id,
				body: 'error'
			});
			return;
		}
 
		cb({
			serverId: agent.id,
			body: connection.getStatisticsInfo()
		});
	}
}
 
 
function showLogins(handle, agent, app, comd, context, cb) {
	showConnections(handle, agent, app, comd, context, cb);
}
 
function showModules(handle, agent, comd, context, cb) {
	var modules = agent.consoleService.modules;
	var result = [];
	for (var module in modules) {
		result.push(module);
	}
	cb(null, {
		msg: result
	});
}
 
function showStatus(handle, agent, comd, context, cb) {
	if (handle === 'client') {
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			context: context
		}, function(err, msg) {
			cb(null, {
				msg: msg
			});
		});
	} else if (handle === 'monitor') {
		var serverId = agent.id;
		var pid = process.pid;
		var params = {
			serverId: serverId,
			pid: pid
		};
		monitor.psmonitor.getPsInfo(params, function(err, data) {
			cb(null, {
				serverId: agent.id,
				body: data
			})
		});
	}
}
 
function showConfig(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (param === 'master') {
			cb(null, {
				masterConfig: app.get('masterConfig') || 'no config to master in app.js',
				masterInfo: app.get('master')
			});
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var key = param + 'Config';
		cb(null, clone(param, app.get(key)));
	}
}
 
function showProxy(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		proxyCb(app, context, cb);
	}
}
 
function showHandler(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		handlerCb(app, context, cb);
	}
}
 
function showComponents(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var _components = app.components;
		var res = {};
		for (var key in _components) {
			var name = getComponentName(key);
			res[name] = clone(name, app.get(name + 'Config'))
		}
		cb(null, res);
	}
}
 
function showSettings(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var _settings = app.settings;
		var res = {};
		for (var key in _settings) {
			if (key.match(/^__\w+__$/) || key.match(/\w+Config$/)) {
				continue;
			}
			if (!checkJSON(_settings[key])) {
				res[key] = 'Object';
				continue;
			}
			res[key] = _settings[key];
		}
		cb(null, res);
	}
}
 
function dumpCPU(handle, agent, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(err, msg);
		});
	} else if (handle === 'monitor') {
		var times = param['times'];
		var filepath = param['filepath'];
		var force = param['force'];
		cb(null, 'cpu dump is unused in 1.0 of pomelo');
		/**
		if (!/\.cpuprofile$/.test(filepath)) {
			filepath = filepath + '.cpuprofile';
		}
		if (!times || !/^[0-9]*[1-9][0-9]*$/.test(times)) {
			cb('no times or times invalid error');
			return;
		}
		checkFilePath(filepath, force, function(err) {
			if (err) {
				cb(err);
				return;
			}
			//ndump.cpu(filepath, times);
			cb(null, filepath + ' cpu dump ok');
		});
		*/
 
	}
}
 
function dumpMemory(handle, agent, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(err, msg);
		});
	} else if (handle === 'monitor') {
		var filepath = param['filepath'];
		var force = param['force'];
		if (!/\.heapsnapshot$/.test(filepath)) {
			filepath = filepath + '.heapsnapshot';
		}
		checkFilePath(filepath, force, function(err) {
			if (err) {
				cb(err);
				return;
			}
			var heapdump = null;
			try {
				heapdump = require('heapdump');
				heapdump.writeSnapshot(filepath);
				cb(null, filepath + ' memory dump ok')
			} catch (e) {
				cb('pomelo-admin require heapdump');
			}
		});
	}
}
 
function getApp(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var res = app.get(param);
		if (!checkJSON(res)) {
			res = 'object';
		}
		cb(null, res || null);
	}
}
 
function setApp(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var key = param['key'];
		var value = param['value'];
		app.set(key, value);
		cb(null, 'set ' + key + ':' + value + ' ok');
	}
}
 
function enableApp(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		app.enable(param);
		cb(null, 'enable ' + param + ' ok');
	}
}
 
function disableApp(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		app.disable(param);
		cb(null, 'disable ' + param + ' ok');
	}
}
 
function runScript(handle, agent, app, comd, context, param, cb) {
	if (handle === 'client') {
		if (context === 'all') {
			cb('context error');
			return;
		}
 
		agent.request(context, module.exports.moduleId, {
			comd: comd,
			param: param,
			context: context
		}, function(err, msg) {
			cb(null, msg);
		});
	} else if (handle === 'monitor') {
		var ctx = {
			app: app,
			result: null
		};
		try {
			vm.runInNewContext('result = ' + param, ctx, 'myApp.vm');
			cb(null, util.inspect(ctx.result));
		} catch (e) {
			cb(null, e.stack);
		}
	}
}
 
function showError(handle, agent, comd, context, cb) {
 
}
 
function clone(param, obj) {
	var result = {};
	var flag = 1;
	for (var key in obj) {
		if (typeof obj[key] === 'function' || typeof obj[key] === 'object') {
			continue;
		}
		flag = 0;
		result[key] = obj[key];
	}
	if (flag) {
		// return 'no ' + param + 'Config info';
	}
	return result;
}
 
function checkFilePath(filepath, force, cb) {
	if (!force && fs.existsSync(filepath)) {
		cb('filepath file exist');
		return;
	}
	fs.writeFile(filepath, 'test', function(err) {
		if (err) {
			cb('filepath invalid error');
			return;
		}
		fs.unlinkSync(filepath);
		cb(null);
	})
}
 
function proxyCb(app, context, cb) {
	var msg = {};
	var __proxy__ = app.components.__proxy__;
	if (__proxy__ && __proxy__.client && __proxy__.client.proxies.user) {
		var proxies = __proxy__.client.proxies.user;
		var server = app.getServerById(context);
		if (!server) {
			cb('no server with this id ' + context);
		} else {
			var type = server['serverType'];
			var tmp = proxies[type];
			msg[type] = {};
			for (var _proxy in tmp) {
				var r = tmp[_proxy];
				msg[type][_proxy] = {};
				for (var _rpc in r) {
					if (typeof r[_rpc] === 'function') {
						msg[type][_proxy][_rpc] = 'function';
					}
				}
			}
			cb(null, msg);
		}
	} else {
		cb('no proxy loaded');
	}
}
 
function handlerCb(app, context, cb) {
	var msg = {};
	var __server__ = app.components.__server__;
	if (__server__ && __server__.server && __server__.server.handlerService.handlers) {
		var handles = __server__.server.handlerService.handlers;
		var server = app.getServerById(context);
		if (!server) {
			cb('no server with this id ' + context);
		} else {
			var type = server['serverType'];
			var tmp = handles;
			msg[type] = {};
			for (var _p in tmp) {
				var r = tmp[_p];
				msg[type][_p] = {};
				for (var _r in r) {
					if (typeof r[_r] === 'function') {
						msg[type][_p][_r] = 'function';
					}
				}
			}
			cb(null, msg);
		}
	} else {
		cb('no handler loaded');
	}
}
 
function getComponentName(c) {
	var t = c.match(/^__(\w+)__$/);
	if (t) {
		t = t[1];
	}
	return t;
}
 
function checkJSON(obj) {
	if (!obj) {
		return true;
	}
	try {
		JSON.stringify(obj);
	} catch (e) {
		return false;
	}
	return true;
}
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
1
1
1
1
1
1
 
1
1
1
1
1
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
1
 
 
 
1
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'MonitorAgent');
var MqttClient = require('../protocol/mqtt/mqttClient');
var EventEmitter = require('events').EventEmitter;
var protocol = require('../util/protocol');
var utils = require('../util/utils');
var Util = require('util');
 
var ST_INITED = 1;
var ST_CONNECTED = 2;
var ST_REGISTERED = 3;
var ST_CLOSED = 4;
var STATUS_INTERVAL = 5 * 1000; // 60 seconds
 
/**
 * MonitorAgent Constructor
 *
 * @class MasterAgent
 * @constructor
 * @param {Object} opts construct parameter
 *                 opts.consoleService {Object} consoleService
 *                 opts.id             {String} server id
 *                 opts.type           {String} server type, 'master', 'connector', etc.
 *                 opts.info           {Object} more server info for current server, {id, serverType, host, port}
 * @api public
 */
var MonitorAgent = function(opts) {
  EventEmitter.call(this);
  this.reqId = 1;
  this.opts = opts;
  this.id = opts.id;
  this.socket = null;
  this.callbacks = {};
  this.type = opts.type;
  this.info = opts.info;
  this.state = ST_INITED;
  this.consoleService = opts.consoleService;
};
 
Util.inherits(MonitorAgent, EventEmitter);
 
/**
 * register and connect to master server
 *
 * @param {String} port
 * @param {String} host
 * @param {Function} cb callback function
 * @api public
 */
MonitorAgent.prototype.connect = function(port, host, cb) {
  if (this.state > ST_INITED) {
    logger.error('monitor client has connected or closed.');
    return;
  }
 
  cb = cb || function() {}
 
  this.socket = new MqttClient(this.opts);
  this.socket.connect(host, port);
 
  // this.socket = sclient.connect(host + ':' + port, {
  //   'force new connection': true,
  //   'reconnect': true,
  //   'max reconnection attempts': 20
  // });
  var self = this;
  this.socket.on('register', function(msg) {
    if (msg && msg.code === protocol.PRO_OK) {
      self.state = ST_REGISTERED;
      cb();
    } else {
      self.emit('close');
      logger.error('server %j %j register master failed', self.id, self.type);
    }
  });
 
  this.socket.on('monitor', function(msg) {
    if (self.state !== ST_REGISTERED) {
      return;
    }
 
    msg = protocol.parse(msg);
 
    if (msg.command) {
      // a command from master
      self.consoleService.command(msg.command, msg.moduleId, msg.body, function(err, res) {
        //notify should not have a callback
      });
    } else {
      var respId = msg.respId;
      if (respId) {
        // a response from monitor
        var respCb = self.callbacks[respId];
        if (!respCb) {
          logger.warn('unknown resp id:' + respId);
          return;
        }
        delete self.callbacks[respId];
        respCb(msg.error, msg.body);
        return;
      }
 
      // request from master
      self.consoleService.execute(msg.moduleId, 'monitorHandler', msg.body, function(err, res) {
        if (protocol.isRequest(msg)) {
          var resp = protocol.composeResponse(msg, err, res);
          if (resp) {
            self.doSend('monitor', resp);
          }
        } else {
          //notify should not have a callback
          logger.error('notify should not have a callback.');
        }
      });
    }
  });
 
  this.socket.on('connect', function() {
    if (self.state > ST_INITED) {
      //ignore reconnect
      return;
    }
    self.state = ST_CONNECTED;
    var req = {
      id: self.id,
      type: 'monitor',
      serverType: self.type,
      pid: process.pid,
      info: self.info
    };
    var authServer = self.consoleService.authServer;
    var env = self.consoleService.env;
    authServer(req, env, function(token) {
      req['token'] = token;
      self.doSend('register', req);
    });
  });
 
  this.socket.on('error', function(err) {
    if (self.state < ST_CONNECTED) {
      // error occurs during connecting stage
      cb(err);
    } else {
      self.emit('error', err);
    }
  });
 
  this.socket.on('disconnect', function(reason) {
    self.state = ST_CLOSED;
    self.emit('close');
  });
 
  this.socket.on('reconnect', function() {
    self.state = ST_CONNECTED;
    var req = {
      id: self.id,
      type: 'monitor',
      info: self.info,
      pid: process.pid,
      serverType: self.type
    };
 
    self.doSend('reconnect', req);
  });
 
  this.socket.on('reconnect_ok', function(msg) {
    if (msg && msg.code === protocol.PRO_OK) {
      self.state = ST_REGISTERED;
    }
  });
};
 
/**
 * close monitor agent
 *
 * @api public
 */
MonitorAgent.prototype.close = function() {
  if (this.state >= ST_CLOSED) {
    return;
  }
  this.state = ST_CLOSED;
  this.socket.disconnect();
};
 
/**
 * set module
 *
 * @param {String} moduleId module id/name
 * @param {Object} value module object
 * @api public
 */
MonitorAgent.prototype.set = function(moduleId, value) {
  this.consoleService.set(moduleId, value);
};
 
/**
 * get module
 *
 * @param {String} moduleId module id/name
 * @api public
 */
MonitorAgent.prototype.get = function(moduleId) {
  return this.consoleService.get(moduleId);
};
 
/**
 * notify master server without callback
 *
 * @param {String} moduleId module id/name
 * @param {Object} msg message
 * @api public
 */
MonitorAgent.prototype.notify = function(moduleId, msg) {
  if (this.state !== ST_REGISTERED) {
    logger.error('agent can not notify now, state:' + this.state);
    return;
  }
  this.doSend('monitor', protocol.composeRequest(null, moduleId, msg));
  // this.socket.emit('monitor', protocol.composeRequest(null, moduleId, msg));
};
 
MonitorAgent.prototype.request = function(moduleId, msg, cb) {
  if (this.state !== ST_REGISTERED) {
    logger.error('agent can not request now, state:' + this.state);
    return;
  }
  var reqId = this.reqId++;
  this.callbacks[reqId] = cb;
  this.doSend('monitor', protocol.composeRequest(reqId, moduleId, msg));
  // this.socket.emit('monitor', protocol.composeRequest(reqId, moduleId, msg));
};
 
MonitorAgent.prototype.doSend = function(topic, msg) {
  this.socket.send(topic, msg);
}
 
module.exports = MonitorAgent;
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
1
1
1
1
1
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
1
 
 
 
 
 
1
 
 
 
 
1
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'MqttClient');
var EventEmitter = require('events').EventEmitter;
var constants = require('../../util/constants');
var MqttCon = require('mqtt-connection');
var Util = require('util');
var net = require('net');
 
var MqttClient = function(opts) {
	EventEmitter.call(this);
	this.clientId = 'MQTT_ADMIN_' + Date.now();
	this.id = opts.id;
	this.requests = {};
	this.connectedTimes = 1;
	this.host = null;
	this.port = null;
	this.socket = null;
	this.lastPing = -1;
	this.lastPong = -1;
	this.closed = false;
	this.timeoutId = null;
	this.connected = false;
	this.reconnectId = null;
	this.timeoutFlag = false;
	this.keepaliveTimer = null;
	this.reconnectDelay = 0;
	this.reconnectDelayMax = opts.reconnectDelayMax || constants.DEFAULT_PARAM.RECONNECT_DELAY_MAX;
	this.timeout = opts.timeout || constants.DEFAULT_PARAM.TIMEOUT;
	this.keepalive = opts.keepalive || constants.DEFAULT_PARAM.KEEPALIVE;
}
 
Util.inherits(MqttClient, EventEmitter);
 
MqttClient.prototype.connect = function(host, port, cb) {
	cb = cb || function() {}
	if (this.connected) {
		return cb(new Error('MqttClient has already connected.'));
	}
 
	if (host) {
		this.host = host;
	} else {
		host = this.host;
	}
 
	if (port) {
		this.port = port;
	} else {
		port = this.port;
	}
 
	var self = this;
	this.closed = false;
 
	var stream = net.createConnection(this.port, this.host);
	this.socket = MqttCon(stream);
 
	// logger.info('try to connect %s %s', this.host, this.port);
	this.socket.connect({
		clientId: this.clientId
	});
 
	this.addTimeout();
 
	this.socket.on('connack', function() {
		if (self.connected) {
			return;
		}
 
		self.connected = true;
 
		self.setupKeepAlive();
 
		if (self.connectedTimes++ == 1) {
			self.emit('connect');
			cb();
		} else {
			self.emit('reconnect');
		}
	});
 
	this.socket.on('publish', function(pkg) {
		var topic = pkg.topic;
		var msg = pkg.payload.toString();
		msg = JSON.parse(msg);
 
		// logger.debug('[MqttClient] publish %s %j', topic, msg);
		self.emit(topic, msg);
	});
 
	this.socket.on('close', function() {
		logger.error('mqtt socket is close, remote server host: %s, port: %s', host, port);
		self.onSocketClose();
	});
 
	this.socket.on('error', function(err) {
		logger.error('mqtt socket is error, remote server host: %s, port: %s', host, port);
		// self.emit('error', new Error('[MqttClient] socket is error, remote server ' + host + ':' + port));
		self.onSocketClose();
	});
 
	this.socket.on('pingresp', function() {
		self.lastPong = Date.now();
	});
 
	this.socket.on('disconnect', function() {
		logger.error('mqtt socket is disconnect, remote server host: %s, port: %s', host, port);
		self.emit('disconnect', self.id);
		self.onSocketClose();
	});
 
	this.socket.on('timeout', function(reconnectFlag) {
		if (reconnectFlag) {
			self.reconnect();
		} else {
			self.exit();
		}
	})
}
 
MqttClient.prototype.send = function(topic, msg) {
	// console.log('MqttClient send %s %j ~~~', topic, msg);
	this.socket.publish({
		topic: topic,
		payload: JSON.stringify(msg)
	})
}
 
MqttClient.prototype.onSocketClose = function() {
	// console.log('onSocketClose ' + this.closed);
	if (this.closed) {
		return;
	}
 
	clearInterval(this.keepaliveTimer);
	clearTimeout(this.timeoutId);
	this.keepaliveTimer = null;
	this.lastPing = -1;
	this.lastPong = -1;
	this.connected = false;
	this.closed = true;
	delete this.socket;
	this.socket = null;
 
	if (this.connectedTimes > 1) {
		this.reconnect();
	} else {
		this.exit();
	}
}
 
MqttClient.prototype.addTimeout = function(reconnectFlag) {
	var self = this;
	if (this.timeoutFlag) {
		return;
	}
 
	this.timeoutFlag = true;
 
	this.timeoutId = setTimeout(function() {
		self.timeoutFlag = false;
		logger.error('mqtt client connect %s:%d timeout %d s', self.host, self.port, self.timeout / 1000);
		self.socket.emit('timeout', reconnectFlag);
	}, self.timeout);
}
 
MqttClient.prototype.reconnect = function() {
	var delay = this.reconnectDelay * 2 || constants.DEFAULT_PARAM.RECONNECT_DELAY;
	if (delay > this.reconnectDelayMax) {
		delay = this.reconnectDelayMax;
	}
 
	this.reconnectDelay = delay;
 
	var self = this;
 
	// logger.debug('[MqttClient] reconnect %d ...', delay);
	this.reconnectId = setTimeout(function() {
		logger.info('reconnect delay %d s', delay / 1000);
		self.addTimeout(true);
		self.connect();
	}, delay);
}
 
MqttClient.prototype.setupKeepAlive = function() {
	clearTimeout(this.reconnectId);
	clearTimeout(this.timeoutId);
 
	var self = this;
	this.keepaliveTimer = setInterval(function() {
		self.checkKeepAlive();
	}, this.keepalive);
}
 
MqttClient.prototype.checkKeepAlive = function() {
	if (this.closed) {
		return;
	}
 
	var now = Date.now();
	var KEEP_ALIVE_TIMEOUT = this.keepalive * 2;
	if (this.lastPing > 0) {
		if (this.lastPong < this.lastPing) {
			if (now - this.lastPing > KEEP_ALIVE_TIMEOUT) {
				logger.error('mqtt rpc client checkKeepAlive error timeout for %d', KEEP_ALIVE_TIMEOUT);
				this.close();
			}
		} else {
			this.socket.pingreq();
			this.lastPing = Date.now();
		}
	} else {
		this.socket.pingreq();
		this.lastPing = Date.now();
	}
}
 
MqttClient.prototype.disconnect = function() {
	this.close();
}
 
MqttClient.prototype.close = function() {
	this.connected = false;
	this.closed = true;
	this.socket.disconnect();
}
 
MqttClient.prototype.exit = function() {
	logger.info('exit ...');
	process.exit(0);
}
 
module.exports = MqttClient;
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
1
1
1
1
1
 
1
 
1
 
 
 
 
 
1
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
var logger = require('pomelo-logger').getLogger('pomelo-admin', 'MqttServer');
var EventEmitter = require('events').EventEmitter;
var MqttCon = require('mqtt-connection');
var Util = require('util');
var net = require('net');
 
var curId = 1;
 
var MqttServer = function(opts, cb) {
	EventEmitter.call(this);
	this.inited = false;
	this.closed = true;
};
 
Util.inherits(MqttServer, EventEmitter);
 
MqttServer.prototype.listen = function(port) {
	//check status
	if (this.inited) {
		this.cb(new Error('already inited.'));
		return;
	}
 
	this.inited = true;
 
	var self = this;
 
	this.server = new net.Server();
	this.server.listen(port);
 
	logger.info('[MqttServer] listen on %d', port);
 
	this.server.on('listening', this.emit.bind(this, 'listening'));
 
	this.server.on('error', function(err) {
		// logger.error('mqtt server is error: %j', err.stack);
		self.emit('error', err);
	});
 
	this.server.on('connection', function(stream) {
		var socket = MqttCon(stream);
		socket['id'] = curId++;
 
		socket.on('connect', function(pkg) {
			socket.connack({
				returnCode: 0
			});
		});
 
		socket.on('publish', function(pkg) {
			var topic = pkg.topic;
			var msg = pkg.payload.toString();
			msg = JSON.parse(msg);
 
			// logger.debug('[MqttServer] publish %s %j', topic, msg);
			socket.emit(topic, msg);
		});
 
		socket.on('pingreq', function() {
			socket.pingresp();
		});
 
		socket.send = function(topic, msg) {
			socket.publish({
				topic: topic,
				payload: JSON.stringify(msg)
			});
		};
 
		self.emit('connection', socket);
	});
};
 
MqttServer.prototype.send = function(topic, msg) {
	this.socket.publish({
		topic: topic,
		payload: msg
	});
}
 
MqttServer.prototype.close = function() {
	if (this.closed) {
		return;
	}
 
	this.closed = true;
	this.server.close();
	this.emit('closed');
};
 
module.exports = MqttServer;
 

1
2
3
4
5
6
7
8
9
10
11
1
 
 
 
 
 
 
 
 
 
 
module.exports = {
	DEFAULT_PARAM: {
		KEEPALIVE: 60 * 1000,
		TIMEOUT: 5 * 1000,
		RECONNECT_DELAY: 1 * 1000,
		RECONNECT_DELAY_MAX: 60 * 1000
	},
	TYPE_CLIENT: 'client',
	TYPE_MONITOR: 'monitor'
};
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
1
 
 
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
var exp = module.exports;
 
/**
 * Count down to zero and invoke cb finally.
 */
var CountDownLatch = function(count, cb) {
  this.count = count;
  this.cb = cb;
};
 
/**
 * Call when a task finish to count down.
 *
 * @api public
 */
CountDownLatch.prototype.done = function() {
  if(this.count <= 0) {
    throw new Error('illegal state.');
  }
 
  this.count--;
  if (this.count === 0) {
    this.cb();
  }
};
 
/**
 * create a count down latch
 *
 * @api public
 */
exp.createCountDownLatch = function(count, cb) {
  if(!count || count <= 0) {
    throw new Error('count should be positive.');
  }
  if(typeof cb !== 'function') {
    throw new Error('cb should be a function.');
  }
 
  return new CountDownLatch(count, cb);
};
 
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
1
 
1
1
 
1
 
 
 
 
 
 
 
 
1
 
1
 
1
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
1
 
 
 
1
 
 
 
1
 
 
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
var fs = require('fs');
 
var HeapProfileType = 'HEAP';
var CPUProfileType = 'CPU';
 
var Proxy = function(){
	this.profiles = {
		HEAP: {},
		CPU: {}
	};
 
	this.isProfilingCPU = false;
};
 
module.exports = Proxy;
 
var pro = Proxy.prototype;
 
pro.enable = function(id, params, clientId, agent) {
	this.sendResult(id,{
		result : true
	}, clientId, agent);
};
 
pro.causesRecompilation = function(id, params, clientId, agent) {
	this.sendResult(id,{
		result: false
	}, clientId, agent);
};
 
pro.isSampling = function(id, params, clientId, agent) {
	this.sendResult(id,{
		result: true
	}, clientId, agent);
};
 
pro.hasHeapProfiler = function(id, params, clientId, agent) {
	this.sendResult(id,{
		result: true
	}, clientId, agent);
};
 
pro.getProfileHeaders = function(id, params, clientId, agent) {
	var headers = [];
	for (var type in this.profiles) {
		for (var profileId in this.profiles[type]) {
			var profile = this.profiles[type][profileId];
			headers.push({
				title: profile.title,
				uid: profile.uid,
				typeId: type
			});
		}
	}
	this.sendResult(id, {
		headers: headers
	}, clientId, agent);
};
 
pro.takeHeapSnapshot = function(id, params, clientId, agent) {
	var uid = params.uid;
 
	agent.notifyById(uid, 'profiler', {type: 'heap', action: 'start', uid: uid, clientId: clientId});
 
	this.sendEvent({
		method: 'Profiler.addProfileHeader', 
		params: {header: {title: uid, uid: uid, typeId: HeapProfileType}}
	}, clientId, agent);
	this.sendResult(id, {}, clientId, agent);
};
 
pro.takeSnapCallBack = function (data) {
	var uid = data.params.uid || 0;
	var snapShot = this.profiles[HeapProfileType][uid];
	if (!snapShot || snapShot.finish) {
		snapShot = {};
		snapShot.data = [];
		snapShot.finish = false;
		snapShot.uid = uid;
		snapShot.title = uid;
	}
	if (data.method === 'Profiler.addHeapSnapshotChunk') {
		var chunk = data.params.chunk;
		snapShot.data.push(chunk);
	} else {
		snapShot.finish = true;
	}
	this.profiles[HeapProfileType][uid] = snapShot;
};
 
pro.getProfile = function(id, params, clientId, agent) {
	var profile = this.profiles[params.type][params.uid];
	var self = this;
	if (!profile || !profile.finish) {
		var timerId = setInterval(function() {
			profile = self.profiles[params.type][params.uid];
			if (!!profile) {
				clearInterval(timerId);
				self.asyncGet(id, params, profile, clientId, agent);
			}
		}, 5000);
	} else {
		this.asyncGet(id,params, profile, clientId, agent);
	}
};
 
pro.asyncGet = function(id, params, snapshot, clientId, agent) {
	var uid = params.uid;
	if (params.type === HeapProfileType) {
		for (var index in snapshot.data) {
			var chunk = snapshot.data[index];
			this.sendEvent({method: 'Profiler.addHeapSnapshotChunk', params: {uid: uid, chunk: chunk}}, clientId, agent);
		}
		this.sendEvent({method: 'Profiler.finishHeapSnapshot', params: {uid: uid}}, clientId, agent);
		this.sendResult(id, {profile: {title: snapshot.title, uid: uid, typeId: HeapProfileType}}, clientId, agent);
	} else if (params.type === CPUProfileType) {
		this.sendResult(id,{
			profile: {
				title: snapshot.title,
				uid: uid,
				typeId: CPUProfileType,
				head: snapshot.data.head,
				bottomUpHead: snapshot.data.bottomUpHead
			}
		}, clientId, agent);
	}
};
 
pro.clearProfiles = function(id, params) {
	this.profiles.HEAP = {};
	this.profiles.CPU = {};
	//profiler.deleteAllSnapshots();
	//profiler.deleteAllProfiles();
};
 
pro.sendResult = function(id, res, clientId, agent){
	agent.notifyClient(clientId, 'profiler', JSON.stringify({id: id, result: res}));
};
 
pro.sendEvent = function(res, clientId, agent){
	agent.notifyClient(clientId, 'profiler', JSON.stringify(res));
};
 
pro.start = function(id, params, clientId, agent) {
	var uid = params.uid;
 
	agent.notifyById(uid, 'profiler', {type: 'CPU', action: 'start', uid: uid, clientId: clientId});
	this.sendEvent({method: 'Profiler.setRecordingProfile', params: {isProfiling: true}}, clientId, agent);
	this.sendResult(id, {}, clientId, agent);
};
 
pro.stop = function(id, params, clientId, agent) {
	var uid = params.uid;
	agent.notifyById(uid, 'profiler', {type: 'CPU', action: 'stop', uid: uid, clientId: clientId});
	this.sendResult(id, {}, clientId, agent);
};
 
pro.stopCallBack = function(res, clientId, agent) {
	var uid = res.msg.uid;
	var profiler = this.profiles[CPUProfileType][uid];
	if (!profiler || profiler.finish){
		profiler = {};
		profiler.data = null;
		profiler.finish = true;
		profiler.typeId = CPUProfileType;
		profiler.uid = uid;
		profiler.title = uid;
	}
	profiler.data = res;
	this.profiles[CPUProfileType][uid] = profiler;
	this.sendEvent({
		method: 'Profiler.addProfileHeader', 
		params: {header: {title: profiler.title, uid: uid, typeId: CPUProfileType}}
	}, clientId, agent);
};